Skip to content

测试平台开发

课程目标

  • 测试平台的价值体系
  • 测试平台的学习路线
  • 测试平台实战练习

问题思考

  1. 为什么要做测试平台？为什么我们要学习平台开发的技术？怎么样去做测试平台才能达到价值最大化（个人、公司）
  2. 平台开发技术要学到什么程度？一定要非常深入吗？
  3. 平台开发需要掌握什么技术栈，如何学习可以提升投入产出比？

测试平台-功能设计

功能点 描述 案例
项目管理 对于项目的分类，通常以业务线为维度 商城、飞书、企业微信
测试类型 手工测试、自动化测试 pytest、JUnit、HttpRunner、Postman
测试用例 测试用例的信息 标题、步骤、预期结果、实际结果
测试套件 用例的集合 通常是用例从不同维度组合，冒烟测试用例集，某业务线用例集
测试装置 用例执行前后的准备工作，处理工作 用例依赖环境的安装，比如 python requirements
测试 Runner 测试用例执行器调度器 执行环境的管理：性能测试环境、兼容性环境
测试报告 测试结果的体现 allure 报告、手工报告

测试平台的个人价值

  • 个人价值：
    • 测试方案能力。（重要）
    • 测试工具产品设计能力。（重要）
    • 前后端的技术能力。

测试平台的产品价值

  • 公司收益（提效）：

    1. 提效
    2. 提升质量
    3. 降低成本
  • 市场收益：
    1. 知名度与影响力
    2. 成熟的商业模式，也有巨大的市场空间。带来直接的创收。
    3. 代表产品： wetest、禅道、process on、figma

测试平台的“Timing”

  1. 需求： 公司需要测试能力服务化、平台化，比如整合内部的多套平台。
  2. 体系： 公司有成熟的测试体系：自动化测试、持续集成
  3. 人力： 有成熟的技术团队，包括前端、后端、测试开发三个角色

测试平台的学习路线

uml diagram

技术-前端

  • 界面展示
  • 良好的用户体验
  • 数据展示
  • 相关技术： VUE、React 等

技术-后端

  • 把控业务逻辑
  • 数据库交互
  • 相关技术：Flask、Django、SpringBoot、Fastapi 等

数据库

  • 数据的持久化存储
  • 数据操作
  • 相关技术： MySQL、Oracle、MongoDB、Redis 等

测试用例管理

  • 测试标题
  • 测试套件
  • 测试步骤
  • 测试装置
  • 测试断言
  • 测试结果
  • 测试执行器

参考知识 xUnit：https://en.wikipedia.org/wiki/XUnit

测试用例调度

  • 定时任务
  • 分布式
  • 触发机制
  • 拓展性配置
  • 相关技术： Jenkins

不分层后端代码的问题

如果一家饭店，从服务员->厨师->收银都是一个人，会成什么样子？

  • 耦合性强
  • 可维护性差
  • 可测性差

代码分层的目的

各司其职（高内聚），轻松协作（低耦合），就是分层思想的目标。

  • 方便后续代码进行维护扩展。
  • 分层的效果需要让整个团队都接受
  • 各个层职责边界清晰

代码分层架构

  • 路由层(Controller)
  • 服务层(Service)
  • 数据层(Dao)
  • Model 层（实体类）

uml diagram

测试平台技术架构

uml diagram

实战1：架构优化

  • 按照标准的设计，完成目录结构优化

uml diagram

实体类

  • Model实体类 的作用一般是和数据表做映射
  • 此对象与数据库表结构一一对应

Model层创建-测试用例表

# 创建用例表
class TestcaseModel(Base):
    # 表名
    __tablename__ = "testcase"
    # 用例id，主键，唯一
    id = Column(Integer, primary_key=True)
    # 用例标题，不为空，并且唯一
    name = Column(String(80), nullable=False, unique=True)
    # 用例步骤
    step = Column(String(120))
    # 用例的自动化方法 => CalculatorProject/test/cases/test_add.py
    method = Column(String(120))
    # 备注
    remark = Column(String(120))

    def __repr__(self):
        # 数据库的 魔法方法 直观展示数据
        '''[<User "xxxx">,<User "yyyy">]'''
        return '<Testcase %r>' % self.name

    def as_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "step": self.step,
            "method": self.method,
            "remark": self.remark,
        }

Model层创建-测试计划表

# 创建测试计划表
class PlanModel(Base):
    # 表名
    __tablename__ = "plan"
    # 测试计划 ID
    id = Column(Integer, primary_key=True)
    # 测试计划名称
    name = Column(String(80), nullable=False, unique=True)
    # 测试用例列表
    testcases = relationship("TestcaseModel", secondary=testcase_plan_rel, backref='plan')

    def __repr__(self):
        # 数据库的 魔法方法 直观展示数据
        '''[<User "xxxx">,<User "yyyy">]'''
        return '<Plan %r>' % self.name

    def as_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "testcases": [testcase.as_dict() for testcase in self.testcases],
        }

Model层创建-构建记录表

# 创建构建记录表
class RecordModel(Base):
    __tablename__ = "record"

    # 在新系统下，使用带有显式注释的Mypy插件而在注释中不使用Mapped的SQLAlchemy应用程序会出现错误，
    # 因为在使用relationship（）等结构时，这些注释会被标记为错误。
    # 将__allow_unmapped__添加到显式类型的ORM模型一节说明了如何暂时禁止为使用显式注释的遗留ORM模型引发这些错误。
    __allow_unmapped__ = True

    # 执行记录 ID
    id = Column(Integer, primary_key=True)
    # 测试计划 ID
    plan_id = Column(Integer, ForeignKey('plan.id', ondelete='CASCADE'))
    # 测试报告
    report = Column(String(80), nullable=False, unique=True)
    # 执行时间
    create_time = Column(DateTime, nullable=True, default=datetime.now().strftime('%Y-%m-%d %H:%M:%S'))

    # 参数1： 关联的另外一个业务表类名， 参数2： 反射别名
    plan: PlanModel = relationship("PlanModel", backref="record")

    def __repr__(self):
        # 数据库的 魔法方法 直观展示数据
        '''[<User "xxxx">,<User "yyyy">]'''
        return '<Record %r>' % self.id

    def as_dict(self):
        return {
            "id": self.id,
            "plan_id": self.plan_id,
            "report": self.report,
            "create_time": str(self.create_time)
        }

Model层创建-多对多（测试用例与测试计划）

# 中间表
# 中间表表名
# 测试用例的外键
# 计划的外键
testcase_plan_rel = Table(
    "testcase_plan_rel",  # 表名
    Base.metadata,  # 表继承的类
    # 参数一： 表名_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)
)

Model层创建-用户表

class UserModel(Base):
    # 表名
    __tablename__ = "user"

    # 用户 ID, 用户的唯 一标识
    id = Column(Integer, primary_key=True)
    # 用户名, 限定 80个字符 ，不为空，并且唯一
    username = Column(String(80), nullable=False, unique=True)
    # 密码
    password = Column(String(500), nullable=False)
    # 创建时间,不需要手动传入，在写入记录的时候自动生成
    create_time = Column(DateTime, nullable=True, default=datetime.now().strftime('%Y-%m-%d %H:%M:%S'))

    def __init__(self, *args, **kwargs):
        self.username = kwargs.get('username')
        self.password = generate_password_hash(kwargs.get('password'))

    # 在登陆时候的验证操作
    # 这里的参数是hash过的参数以及原始传入hash
    def check_hash_password(self, raw_password):
        password = check_password_hash(self.password, raw_password)
        return password

    def as_dict(self):
        return {
            "id": self.id,
            "username": self.username,
            "password": self.password,
            "create_time": str(self.create_time)
        }

Model层创建-初始化数据库

from server import Base, engine
from model.testcase_model import TestcaseModel
from model.plan_model import PlanModel
from model.record_model import RecordModel
from model.user_model import UserModel

if __name__ == '__main__':
    # 删除所有数据
    # Base.metadata.drop_all(bind=engine)
    # 创建表，需要传入创建连接的对象
    Base.metadata.create_all(bind=engine)

实战2：创建表结构

完成测试计划、测试用例、构建记录的表创建

Dao层实现

  • 数据库增删查改动作封装

Dao层创建-测试用例

# Dao 负责和数据库的交互
class TestcaseDao:

    def get(self, testcase_id: int) -> TestcaseModel:
        """
        添加用例
        :param testcase_id: 用例id
        :return: TestcaseModel
        """
        return db_session.query(TestcaseModel).filter_by(id=testcase_id).first()

    def get_by_name(self, testcase_name: str) -> TestcaseModel:
        """
        根据测试用例名称查询
        """
        return db_session.query(TestcaseModel).filter_by(name=testcase_name).first()

    def list(self):
        """
        获取用例列表
        :return:
        """
        return db_session.query(TestcaseModel).all()

    def create(self, testcase_model: TestcaseModel) -> int:
        """
        创建用例
        :param testcase_model: testcase对象
        :return:
        """
        db_session.add(testcase_model)
        db_session.commit()
        return testcase_model.id

    def delete(self, testcase_id: int) -> int:
        """
        删除用例
        :param testcase_id: 用例id
        :return:
        """
        db_session.query(TestcaseModel).filter_by(id=testcase_id).delete()
        db_session.commit()
        return testcase_id

    def update(self, testcase_model: TestcaseModel) -> int:
        """
        更新用例
        :param testcase_model: testcase对象
        :param testcase_id: 用例id
        :return:
        """
        db_session.query(TestcaseModel).filter_by(id=testcase_model.id).update(testcase_model.as_dict())
        db_session.commit()
        return testcase_model.id

Dao层创建-测试计划

# Dao 负责和数据库的交互
class PlanDao:

    def get(self, plan_id) -> PlanModel:
        # 根据id返回数据
        return db_session.query(PlanModel).filter_by(id=plan_id).first()

    def get_by_name(self, name) -> PlanModel:
        # 根据name返回数据
        return db_session.query(PlanModel).filter_by(name=name).first()

    def list(self):
        # 返回所有数据
        return db_session.query(PlanModel).all()

    def create(self, plan_model: PlanModel):
        # 新增数据
        db_session.add(plan_model)
        db_session.commit()
        return plan_model.id

    def delete(self, plan_id):
        # 删除操作
        db_session.query(PlanModel).filter_by(id=plan_id).delete()
        db_session.commit()
        return plan_id

Dao层创建-构建记录

# Dao 负责和数据库的交互
class RecordDao:

    def list_by_plan_id(self, plan_id):
        # 根据id返回数据
        return db_session.query(RecordModel).filter_by(plan_id=plan_id).all()

    def list(self):
        # 返回所有数据
        return db_session.query(RecordModel).all()

    def create(self, record_model: RecordModel):
        # 新增数据
        db_session.add(record_model)
        db_session.commit()
        return record_model.id

Dao层创建-用户

class UserDao:

    def get(self, user_id) -> UserModel:
        '''
        根据 ID 查询用户
        '''
        return db_session.query(UserModel).filter_by(id=user_id).first()

    def get_by_name(self, user_name) -> UserModel:
        '''
        根据姓名查询用户
        '''
        return db_session.query(UserModel).filter_by(username=user_name).first()

    def list(self):
        """
        获取用户列表
        :return:
        """
        return db_session.query(UserModel).all()

    def create(self, user_model: UserModel):
        '''
        创建用户
        '''
        db_session.add(user_model)
        db_session.commit()
        return user_model.id

实战3：完成Dao层逻辑

完成测试计划、测试用例、构建记录的Dao层代码

Server层实现

  • 处理核心的业务逻辑

Server层创建-测试用例

testcase_dao = TestcaseDao()


class TestcaseService:
    def create(self, testcase_model: TestcaseModel) -> int:
        """
        创建用例
        """
        result = testcase_dao.get_by_name(testcase_model.name)
        if not result:
            return testcase_dao.create(testcase_model)

    def update(self, testcase_model: TestcaseModel) -> int:
        """
        更新用例
        """
        if testcase_dao.get(testcase_model.id):
            testcase_dao.update(testcase_model)
        return testcase_model.id

    def delete(self, testcase_id: int) -> int:
        """
        删除用例
        """
        if self.get(testcase_id):
            return testcase_dao.delete(testcase_id)

    def list(self) -> List[TestcaseModel]:
        """
        获取全部用例
        """
        return testcase_dao.list()

    def get(self, testcase_id: int) -> TestcaseModel:
        """
        获取某个测试用例
        """
        return testcase_dao.get(testcase_id)

    def get_by_name(self, testcase_name) -> TestcaseModel:
        """
        根据名称获取某个测试用例
        """
        return testcase_dao.get_by_name(testcase_name=testcase_name)

Server层创建-测试计划

plan_dao = PlanDao()
testcase_service = TestcaseService()


class PlanService:

    def get(self, plan_id):
        '''
        通过 ID 查询测试计划
        '''
        return plan_dao.get(plan_id)

    def get_by_name(self, plan_name):
        '''
        通过名称查询测试计划
        '''
        return plan_dao.get_by_name(plan_name)

    def list(self):
        '''
        返回所有测试计划
        '''
        return plan_dao.list()

    def create(self, plan_model: PlanModel, testcase_id_list):
        '''
        创建测试计划
        '''
        # 创建之前先通过名称查询计划是否已经存在
        plan = self.get_by_name(plan_model.name)
        print(f"名称 {plan_model.name} 的查询结果为 {plan}")
        # 不存在则新增
        if not plan:
            # 根据测试用例 ID 查询获取测试用例对象列表 [1,2,3,4] => [TestcaseModel, TestcaseModel, TestcaseModel]
            testcase_list = [testcase_service.get(testcase_id) for testcase_id in testcase_id_list]
            # 构建测试计划对象
            plan_model.testcases = testcase_list
            # 创建测试计划
            return plan_dao.create(plan_model)
        # 存在则返回 False
        return False

    def delete(self, plan_id):
        # 删除操作
        # 删除之前先查询数据是否存在，存在则进行删除，不存在则返回false
        plan = self.get(plan_id)
        if not plan:
            return False
        else:
            return plan_dao.delete(plan_id)

Server层创建-构建记录

record_dao = RecordDao()
plan_service = PlanService()


class RecordService:

    def list_by_plan(self, plan_id):
        '''
        根据测试计划 ID 获取对应的测试记录
        '''
        return record_dao.list_by_plan_id(plan_id)

    def list(self):
        '''
        获取所有执行记录
        '''
        return record_dao.list()

    def create(self, plan_id):
        '''
        新增测试记录
        '''
        # 新增之前先查询要执行的测试计划是否存在
        plan = plan_service.get(plan_id)
        # 不存在则返回 False
        if not plan:
            return False
        # 存在则创建测试记录
        else:
            # 执行命令格式
            # pytest test_demo1 test_demo2 test_demo3
            # 获取测试记录中包含的测试用例的执行方式
            # [CalculatorProject/test/cases/test_add1.py CalculatorProject/test/cases/test_add2.py CalculatorProject/test/cases/test_add3.py]
            method_list = [testcase.method for testcase in plan.testcases]
            # 为测试执行方式列表去重
            methods = set(method_list)
            print(f"去重后的 method 列表为 {methods}")
            # 获取测试记录中包含的测试用例 step 中包含的 nodeid
            # test_demo1 ==> test_add_params.py
            nodeid_list = [testcase.step for testcase in plan.testcases]
            nodeid = " ".join(nodeid_list)
            print(f"获取到的用例 id 列表为 {nodeid}")
            # 执行测试用例
            invoke_params = {
                "methods": 'pytest',
                "steps": nodeid
            }
            report = JenkinsUtils.invoke(invoke_params)
            # 构造测试记录对象
            record_model = RecordModel(plan_id=plan_id, report=report)
            # 新增测试记录
            return record_dao.create(record_model)

Server层创建-用户

user_dao = UserDao()


class UserService:

    def get(self, user_id) -> UserModel:
        '''
        通过 ID 查询用户
        '''
        return user_dao.get(user_id)

    def get_by_name(self, user_name):
        '''
        通过姓名查询用户
        '''
        return user_dao.get_by_name(user_name)

    def list(self):
        '''
        通过所有用户列表
        '''
        return user_dao.list()

    def create(self, user_model: UserModel) -> int:
        '''
        创建用户
        '''
        # 新增前先查询用户是否存在
        user = user_dao.get_by_name(user_model.username)
        if not user:
            # 没有重名，创建用户
            user_dao.create(user_model)
            return user_model.id

    def create_access_token(self, user_model):
        '''
        根据用户信息生成 token
        '''
        # 使用 jwt 生成 token
        with app.app_context():
            # token = create_access_token(identity=user_model.username, expires_delta=timedelta(days=1))
            # identity 是生成 token 的依据，需要 json 格式的可序列化数据
            # expires_delta 可以配置 token 的过期时间
            token = create_access_token(
                identity=user_model.as_dict(),
                expires_delta=timedelta(days=1)
            )
            return token

    # 配置回调函数中验证数据条件
    @jwt.user_lookup_loader
    def user_lookup_callback(self, _jwt_header, jwt_data):
        # 因为 sub 字段下是请求 body 的内容，所以从 sub 中获取用户名用来查库
        # 获取 username
        username = jwt_data["sub"]["username"]
        # 返回通过 username 查询用户的结果
        return self.get_by_name(username)

工具相关-测试用例调度

class JenkinsUtils:
    jenkins_conf = get_jenkins_config()
    # Jenkins 服务
    BASE_URL = jenkins_conf['url']
    # Jenkins 服务对应的用户名
    USERNAME = jenkins_conf['user']
    # Jenkins 服务对应的token
    PASSWORD = jenkins_conf['password']
    JOB = jenkins_conf['job_name']

    @classmethod
    def invoke(cls, invoke_params):
        """
        执行构建任务
        :return:
        """
        # 获取 Jenkins 对象
        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={
        #     "step": "xx.py::TestXX::test_xx ----",
        #     "methods": "pytest"
        # })
        job.invoke(build_params=invoke_params)
        # job.invoke()
        # 获取job 最后一次完成构建的编号
        last_build_number = job.get_last_buildnumber() + 1
        # 执行方式为：pytest 用例名称 指定报告生成地址
        # 生成报告格式为
        # http://42.192.73.147:7080/job/tpf/22/allure/
        # 获取本次构建的报告地址
        report = f"{cls.BASE_URL}job/{cls.JOB}/{last_build_number}/allure/"
        return report

工具类-配置文件

root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

def get_db_config():
    with open(os.path.join(root_dir, 'conf/conf.yaml'), 'r') as f:
        db_conf = yaml.safe_load(f)['database']
    return db_conf


def get_jenkins_config():
    with open(os.path.join(root_dir, 'conf/conf.yaml'), 'r') as f:
        jenkins_conf = yaml.safe_load(f)['jenkins']
    return jenkins_conf

实战4：测试用例与服务层的实现

完成测试计划、测试用例、构建记录的Service层代码

Controller层实现

  • 处理接口数据、接口文档

Controller层创建-测试用例

case_ns = Namespace("case", description="用例管理")

testcase_service = TestcaseService()


@case_ns.route("")
class TestcaseController(Resource):
    testcase_get_parser = api.parser()
    testcase_get_parser.add_argument("id", type=int, location="args")

    @jwt_required()
    @case_ns.doc(security='Bearer Auth')
    @case_ns.expect(testcase_get_parser)
    def get(self):
        """
        测试用例的查找
        :return:
        """
        id = request.args.get("id")
        if id:
            data = testcase_service.get(int(id))
            if data:
                datas = [testcase_service.get(int(id)).as_dict()]
                return {"code": 0, "msg": "data success get", "data": datas}
            else:
                return {"code": 40004, "msg": "data is not exists"}
        else:
            datas = [testcase.as_dict() for testcase in testcase_service.list()]
            return {"code": 0, "msg": "data success get", "data": datas}

    testcase_post_model = case_ns.model("testcase_post_model", {
        "name": fields.String,
        "step": fields.String,
        "method": fields.String,
        "remark": fields.String
    })

    @jwt_required()
    @case_ns.doc(security='Bearer Auth')
    @case_ns.expect(testcase_post_model)
    def post(self):
        """
        测试用例的新增
        :return:
        """
        data = request.json
        # {a=1,b=2}   a=1,b=2
        testcase = TestcaseModel(**data)
        testcase_id = testcase_service.create(testcase)
        if testcase_id:
            return {"code": 0, "msg": f"{testcase_id} success add"}
        else:
            return {"code": 40001, "msg": "case is exists"}

    testcase_put_model = case_ns.model("testcase_put_model", {
        "id": fields.Integer,
        "name": fields.String,
        "step": fields.String,
        "method": fields.String,
        "remark": fields.String
    })

    @jwt_required()
    @case_ns.doc(security='Bearer Auth')
    @case_ns.expect(testcase_put_model)
    def put(self):
        """
        测试用例的修改
        :return:
        """
        data = request.json
        # {a=1,b=2}   a=1,b=2
        testcase = TestcaseModel(**data)
        testcase_id = testcase_service.update(testcase)
        if testcase_id:
            return {"code": 0, "msg": f"{testcase_id} success update"}
        else:
            return {"code": 40001, "msg": "case is not exists"}

    testcase_delete_parser = api.parser()
    testcase_delete_parser.add_argument("id", type=int, location="json", required=True)

    @jwt_required()
    @case_ns.doc(security='Bearer Auth')
    @case_ns.expect(testcase_delete_parser)
    def delete(self):
        """
        测试用例的删除
        :return:
        """
        data = request.json
        case_id = data.get("id")
        testcase_id = testcase_service.delete(case_id)
        if testcase_id:
            return {"code": 0, "msg": f"{testcase_id} success delete"}
        else:
            return {"code": 40002, "msg": "case is not exists"}

Controller层创建-测试计划

plan_ns = Namespace("plan", description="测试计划管理")

plan_service = PlanService()


@plan_ns.route("")
class PlanController(Resource):
    plan_get_parser = api.parser()
    plan_get_parser.add_argument("id", type=int, location="args")

    @jwt_required()
    @plan_ns.doc(security='Bearer Auth')
    @plan_ns.expect(plan_get_parser)
    def get(self):
        """
        测试计划的查找
        :return:
        """
        id = request.args.get("id")
        print(id)
        if id:
            data = plan_service.get(id)
            if data:
                datas = [plan_service.get(id).as_dict()]
                return {"code": 0, "msg": "data success get", "data": datas}
            else:
                return {"code": 40004, "msg": "data is not exists"}

        else:
            datas = [plan.as_dict() for plan in plan_service.list()]
            return {"code": 0, "msg": "data success get", "data": datas}

    plan_post_model = plan_ns.model("plan_post_model", {
        "name": fields.String,
        "testcase_ids": fields.List(fields.Integer)
    })

    @jwt_required()
    @plan_ns.doc(security='Bearer Auth')
    @plan_ns.expect(plan_post_model)
    def post(self):
        """
        测试计划的新增
        :return:
        """
        data = request.json
        # {a=1,b=2}   a=1,b=2
        testcase_id_list = data.pop("testcase_ids")
        plan = PlanModel(**data)
        plan_id = plan_service.create(plan, testcase_id_list)
        if plan_id:
            return {"code": 0, "msg": f"{plan_id} success add"}
        else:
            return {"code": 40001, "msg": "case is exists"}

    plan_delete_parser = api.parser()
    plan_delete_parser.add_argument("id", type=int, location="json", required=True)

    @jwt_required()
    @plan_ns.doc(security='Bearer Auth')
    @plan_ns.expect(plan_delete_parser)
    def delete(self):
        """
        测试计划的删除
        :return:
        """
        data = request.json
        plan_id = data.get("id")
        id = plan_service.delete(plan_id)
        if id:
            return {"code": 0, "msg": f"{id} success delete"}
        else:
            return {"code": 40002, "msg": "case is not exists"}

Controller层创建-构建记录

record_ns = Namespace("record", description="构建记录管理")

record_service = RecordService()


@record_ns.route("")
class BuildController(Resource):
    build_get_parser = api.parser()
    build_get_parser.add_argument("id", type=int, location="args")

    @jwt_required()
    @record_ns.doc(security='Bearer Auth')
    @record_ns.expect(build_get_parser)
    def get(self):
        """
        构建记录的查找
        :return:
        """
        id = request.args.get("id")
        if id:
            data = record_service.list_by_plan(id)
            if data:
                datas = [record_service.list_by_plan(id).as_dict()]
                return {"code": 0, "msg": "data success get", "data": datas}
            else:
                return {"code": 40004, "msg": "data is not exists"}
        else:
            datas = [plan.as_dict() for plan in record_service.list()]
            return {"code": 0, "msg": "data success get", "data": datas}

    build_post_model = record_ns.model("build_post_model", {
        "plan_id": fields.Integer
    })

    @jwt_required()
    @record_ns.doc(security='Bearer Auth')
    @record_ns.expect(build_post_model)
    def post(self):
        """
        构建记录的新增
        :return:
        """
        plan_id = request.json.get("plan_id")
        build_id = record_service.create(plan_id)
        if build_id:
            return {"code": 0, "msg": f"{build_id} success add"}
        else:
            return {"code": 40001, "msg": "case is exists"}

    build_delete_parser = api.parser()
    build_delete_parser.add_argument("id", type=int, location="json", required=True)

Controller层创建-用户

user_ns = Namespace("user", description="用户管理")

user_service = UserService()


@user_ns.route("/login")
class LoginController(Resource):
    login_post_model = user_ns.model("login_post_model", {
        "username": fields.String,
        "password": fields.String
    })

    @user_ns.expect(login_post_model)
    def post(self):
        """
        登录功能
        :return:
        """
        user_info = request.json
        # 通过用户名和密码生成 user 对象
        user = UserModel(**user_info)
        # 通过用户名查询用户是否存在
        user_result = user_service.get_by_name(user.username)
        # 如果用户不存在，说明用户名未注册
        if not user_result:
            return {"code": 40013, "msg": "用户未注册"}
        # 如果密码不匹配
        if not user_result.check_hash_password(user_info.get("password")):
            # 返回提示信息与未授权的响应状态码
            return {"code": 40014, "msg": "密码错误"}
        # 用户存在，生成 token
        access_token = user_service.create_access_token(user_result)
        if access_token:
            return {"code": 0, "msg": "登录成功", "data": {"token": access_token}}
        else:
            return {"code": 40021, "msg": "登录失败"}


@user_ns.route("/register")
class RegisterController(Resource):
    register_post_model = user_ns.model("register_post_model", {
        "username": fields.String,
        "password": fields.String,
    })

    @user_ns.expect(register_post_model)
    def post(self):
        """
        注册功能
        :return:
        """
        data = request.json
        user = UserModel(**data)
        user_id = user_service.create(user)
        if user_id:
            return {"code": 0, "msg": f"{user_id} success add"}
        else:
            return {"code": 40001, "msg": "user is exists"}


@user_ns.route("/get")
class UserListController(Resource):

    def get(self):
        """
        用户列表
        :return:
        """
        datas = [user.as_dict() for user in user_service.list()]
        return {"code": 0, "msg": "data success get", "data": datas}

启动模块server.py

from flask import Flask
from flask_restx import Api
from flask_cors import CORS
from flask_jwt_extended import JWTManager
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base, scoped_session

from utils.utils import get_db_config

# 定义 app
app = Flask(__name__)
# 解决跨域
CORS(app, supports_credentials=True)
api = Api(app)

# 定义 JWT 安全模式
authorizations = {
    'Bearer Auth': {
        'type': 'apiKey',
        'in': 'header',
        'name': 'Authorization'
    },
}

api.authorizations = authorizations

# 注册 jwt
jwt = JWTManager(app)
# 配置服务端密钥
app.config["JWT_SECRET_KEY"] = "hogwarts_user_AHKFJJD5"
# 开启数据库跟踪模式
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True

# SQLAlchemy 设置
Base = declarative_base()
db_conf = get_db_config()
# 定义数据库
db_user = db_conf["user"]
db_pass = db_conf["password"]
db_host = db_conf["host"]
db_port = db_conf["port"]
db_name = db_conf["db_name"]
# 数据库类型+数据库引擎（ pip install pymysql）
db_url = f'mysql+pymysql://{db_user}:{db_pass}@{db_host}:{db_port}/{db_name}?charset=utf8mb4'
# 创建引擎，连接到数据库
engine = create_engine(db_url, echo=True)
# 创建session对象
# DBSession = sessionmaker(bind=engine)
# db_session: Session = DBSession()
# 解决session的复用问题   不然会报使用的时候前一个session没有回滚
DBSession = scoped_session(sessionmaker(bind=engine))


@app.before_request
def before_request():
    # 在每个请求前执行的代码
    # 在请求开始的时候实例化DBsession
    DBSession()


@app.teardown_request
def teardown_request(exception=None):
    # 在每个请求后执行的代码
    if exception:
        DBSession.rollback()
    # 请求结束之后remove掉DBsession
    DBSession.remove()


def register_router():
    # 如果出现循环导入，把导包语句放在方法内执行。并且调用此函数
    from controller.testcase_controller import case_ns
    from controller.plan_controller import plan_ns
    from controller.record_controller import  record_ns
    from controller.user_controller import user_ns
    api.add_namespace(case_ns, "/testcase")
    api.add_namespace(plan_ns, "/plan")
    api.add_namespace(record_ns, "/build")
    api.add_namespace(user_ns, "/user")



if __name__ == '__main__':
    register_router()
    app.run(debug=True, port=5000)

实战5： 实现接口层逻辑

完成测试计划、测试用例、构建记录的Controller层代码

最终实现效果

总结

  • 项目的整体框架架构
  • Model层、Dao层、Service层、Controller层的实现
  • 项目的启动与接口调用