python自动化自动化测试平台开发：5.后端开发之用例文件下载，执行用例，查询用例执行状态

Posted 2021-06-05 new nm个对象

1.用例文件下载

1.1 接口分析

1.2 实现

第一步：在forms.py文件中，创建用于校验用例文件下载请求参数的校验器

from wtforms import Form, StringField, FileField, IntegerField
from wtforms.validators import Length, InputRequired, ValidationError
from flask_wtf.file import FileRequired


# 校验用例储存请求参数
class TestCaseForm(Form):
    name = StringField(validators=[Length(max=30, message='用例名字不合法，参数长度不能大于30'),
                                   InputRequired(message='用例名字不能为空')
                                   ])
    description = StringField(validators=[Length(max=200, message='用例描述不合法，参数长度不能大于200')])
    test_file = FileField(validators=[FileRequired(message='必须上传用例文件')])

    def validate_test_file(self, field):
        """自定义test_file验证器"""
        if not field.data.filename.endswith('.py'): # field.data获取需要校验的参数值，这里为上传的文件对象
            raise ValidationError('用例文件必须为python文件') # 使用ValidationError抛出提示信息

# 校验用例查看请求参数
class GetTestCaseForm(Form):
    id = IntegerField(validators=[InputRequired('用例id不能为空')])

# 校验用例下载请求参数
class DownloadTestCaseForm(Form):
    id = IntegerField(validators=[InputRequired('用例id不能为空')])

第二步：在app.py文件中增加用例下载的视图函数并映射到url/download_testcase/

import os
from datetime import datetime
from flask import Flask, request, jsonify, send_from_directory
import conf
from flask_restful import Api, Resource
from exit import db
from forms import TestCaseForm, GetTestCaseForm, DownloadTestCaseForm
from models import TestCaseModel
from werkzeug.datastructures import CombinedMultiDict


# 实例化一个Flask对象，用于启动flask服务
app = Flask(__name__)

# 添加配置文件
app.config.from_object(conf)

# 实例化flask_restful对象
api = Api(app)

# 将SQLAlchemy对象db关联到app
db.init_app(app)


# 设置用例储存视图函数
class TestCaseStoreView(Resource):
    def post(self):

        now = datetime.now().strftime('%Y-%m-%d-%H%M%S')  # 生成时间戳
        now_path = os.getcwd()  # 获取当前目录路径

        # 实例化TestCaseForm对象，用于参数校验。其参数为需要校验的参数字典列表
        # CombinedMultiDict将两个属性组合起来传给校验器
        form = TestCaseForm(CombinedMultiDict([request.form,request.files]))


        if form.validate():

            # 参数校验通过
            name = form.name.data # 获取参数name的值
            description = form.description.data # 获取参数description的值
            test_file = form.test_file.data # 获取用例文件test_file对象

            # 判断name是否已在数据库中存在
            name_fag = db.session.query(TestCaseModel).filter(TestCaseModel.name==name).first()
            if name_fag is  None:
                if description is  not None:
                    # 保存用例文件
                    file_path = os.path.join(now_path, f'./case_files/{now}_{test_file.filename}')
                    test_file.save(file_path)

                    # 保存数据库
                    testcase = TestCaseModel(name=name,description=description,file_path=file_path)
                    db.session.add(testcase)
                    db.session.commit()

                    res = {
                        'code': 200,
                        'message': 'success'
                    }
                    return jsonify(res)
                else:
                    # 保存用例文件
                    file_path = os.path.join(now_path, f'./case_files/{now}_{test_file.filename}')
                    test_file.save(file_path)

                    # 保存数据库
                    testcase = TestCaseModel(name=name, file_path=file_path)
                    db.session.add(testcase)
                    db.session.commit()

                    res = {
                        'code': 200,
                        'message': 'success'
                    }
                    return jsonify(res)
            else:
                res = {
                    'code': 404,
                    'message': '用例名字不能重复'
                }
                return jsonify(res)

        else:
            # 参数校验不通过
            res = {
                'code': 404,
                'message': form.errors
            }
            return jsonify(res)

# 查看用例视图函数
class GetTestCaseView(Resource):
    def get(self):
        form = GetTestCaseForm(request.args)
        if form.validate():
            id = form.id.data
            id_fag = db.session.query(TestCaseModel).filter(TestCaseModel.id==id).first()

            # 查看id是否在数据库中存在
            if id_fag is not None:
                res = {
                    'code': 200,
                    'id': id_fag.id,
                    'name': id_fag.name,
                    'description': id_fag.description,
                    'file_path': id_fag.file_path,
                    'update_time': str(id_fag.update_time)
                }
                return jsonify(res)
            else:
                res = {
                    'code': 404,
                    'message': '未找到对应的用例信息'
                }
                return jsonify(res)

        else:
            res = {
                'code':404,
                'message':form.errors
            }
            return jsonify(res)

# 用例文件下载视图函数
class DownloadTestCaseView(Resource):
    def get(self):
        form = DownloadTestCaseForm(request.args)

        if form.validate():
            # 校验通过
            id = form.id.data # 获取请求中的id
            testcase = db.session.query(TestCaseModel).filter(TestCaseModel.id == id).first() # 根据id在数据库中查询用例数据

            if testcase is not None:
                file_path = testcase.file_path
                file_name = file_path.split('/')[-1] # 获取用例文件名

                # 返回用例文件。使用send_from_directory来返回本地的文件，第一个参数为目录，第二个参数为文件名
                # 注意：真正的项目中我们不会将文件储存在本地目录下，而是储存在git或者其他储存服务器中
                file = send_from_directory(f'./case_files',filename=file_name,as_attachment=True)
                return file
            else:
                res = {
                    'code': 404,
                    'message': '未找到对应的用例'
                }
                return jsonify(res)

        else:
            res = {
                'code': 404,
                'message': form.errors
            }
            return jsonify(res)

api.add_resource(TestCaseStoreView, '/testcase_store/', endpoint='testcase_store')
api.add_resource(GetTestCaseView, '/get_testcase/', endpoint='get_testcase')
api.add_resource(DownloadTestCaseView, '/download_testcase/', endpoint='download_testcase')


if __name__ == "__main__":
    app.run(debug=True,host='127.0.0.1',port=5055)

运行结果如下：

2.用例的执行

2.1 接口分析

2.2 接口实现原理

该接口会调用Jenkins的api，执行Jenkins任务。在Jenkins任务中下载用例文件，并执行用例文件。

2.3 实现

（1）创建Jenkins任务，用于执行用例

（2）Jenkinsapi介绍

Jenkinsapi封装了Jenkins的常用操作接口，十分的方便。
官网地址：https://jenkinsapi.readthedocs.io/en/latest/index.html
安装方式：pip install jenkinsapi

（2）完成用例执行接口

第一步：生成token

要使用api去连接Jenkins，需要使用token。所以我们需要先生成一个Jenkins的token。

第二步：完成用例执行的接口功能
先修改forms.py文件，增加一个用于校验用例执行接口请求参数的校验器

from wtforms import Form, StringField, FileField, IntegerField
from wtforms.validators import Length, InputRequired, ValidationError
from flask_wtf.file import FileRequired


# 校验用例储存请求参数
class TestCaseForm(Form):
    name = StringField(validators=[Length(max=30, message='用例名字不合法，参数长度不能大于30'),
                                   InputRequired(message='用例名字不能为空')
                                   ])
    description = StringField(validators=[Length(max=200, message='用例描述不合法，参数长度不能大于200')])
    test_file = FileField(validators=[FileRequired(message='必须上传用例文件')])

    def validate_test_file(self, field):
        """自定义test_file验证器"""
        if not field.data.filename.endswith('.py'): # field.data获取需要校验的参数值，这里为上传的文件对象
            raise ValidationError('用例文件必须为python文件') # 使用ValidationError抛出提示信息

# 校验用例查看请求参数
class GetTestCaseForm(Form):
    id = IntegerField(validators=[InputRequired('用例id不能为空')])

# 校验用例下载请求参数
class DownloadTestCaseForm(Form):
    id = IntegerField(validators=[InputRequired('用例id不能为空')])

# 校验用例执行请求参数
class RunTestCaseForm(Form):
    id = IntegerField(validators=[InputRequired('用例id不能为空')])

修改app.py文件，增加用例执行的视图函数

import os
from datetime import datetime
from flask import Flask, request, jsonify, send_from_directory
import conf
from flask_restful import Api, Resource
from exit import db
from forms import TestCaseForm, GetTestCaseForm, DownloadTestCaseForm, RunTestCaseForm
from models import TestCaseModel
from werkzeug.datastructures import CombinedMultiDict
from jenkinsapi.jenkins import Jenkins
from jenkinsapi import api

# 实例化一个Flask对象，用于启动flask服务
app = Flask(__name__)

# 添加配置文件
app.config.from_object(conf)

# 实例化flask_restful对象
api_ = Api(app)

# 将SQLAlchemy对象db关联到app
db.init_app(app)


# 设置用例储存视图函数
class TestCaseStoreView(Resource):
    def post(self):

        now = datetime.now().strftime('%Y-%m-%d-%H%M%S')  # 生成时间戳
        now_path = os.getcwd()  # 获取当前目录路径

        # 实例化TestCaseForm对象，用于参数校验。其参数为需要校验的参数字典列表
        # CombinedMultiDict将两个属性组合起来传给校验器
        form = TestCaseForm(CombinedMultiDict([request.form,request.files]))


        if form.validate():

            # 参数校验通过
            name = form.name.data # 获取参数name的值
            description = form.description.data # 获取参数description的值
            test_file = form.test_file.data # 获取用例文件test_file对象

            # 判断name是否已在数据库中存在
            name_fag = db.session.query(TestCaseModel).filter(TestCaseModel.name==name).first()
            if name_fag is  None:
                if description is  not None:
                    # 保存用例文件
                    file_path = os.path.join(now_path, f'./case_files/{now}_{test_file.filename}')
                    test_file.save(file_path)

                    # 保存数据库
                    testcase = TestCaseModel(name=name,description=description,file_path=file_path)
                    db.session.add(testcase)
                    db.session.commit()

                    res = {
                        'code': 200,
                        'message': 'success'
                    }
                    return jsonify(res)
                else:
                    # 保存用例文件
                    file_path = os.path.join(now_path, f'./case_files/{now}_{test_file.filename}')
                    test_file.save(file_path)

                    # 保存数据库
                    testcase = TestCaseModel(name=name, file_path=file_path)
                    db.session.add(testcase)
                    db.session.commit()

                    res = {
                        'code': 200,
                        'message': 'success'
                    }
                    return jsonify(res)
            else:
                res = {
                    'code': 404,
                    'message': '用例名字不能重复'
                }
                return jsonify(res)

        else:
            # 参数校验不通过
            res = {
                'code': 404,
                'message': form.errors
            }
            return jsonify(res)

# 查看用例视图函数
class GetTestCaseView(Resource):
    def get(self):
        form = GetTestCaseForm(request.args)
        if form.validate():
            id = form.id.data
            id_fag = db.session.query(TestCaseModel).filter(TestCaseModel.id==id).first()

            # 查看id是否在数据库中存在
            if id_fag is not None:
                res = {
                    'code': 200,
                    'id': id_fag.id,
                    'name': id_fag.name,
                    'description': id_fag.description,
                    'file_path': id_fag.file_path,
                    'update_time': str(id_fag.update_time)
                }
                return jsonify(res)
            else:
                res = {
                    'code': 404,
                    'message': '未找到对应的用例信息'
                }
                return jsonify(res)

        else:
            res = {
                'code':404,
                'message':form.errors
            }
            return jsonify(res)

# 用例文件下载视图函数
class DownloadTestCaseView(Resource):
    def get(self):
        form = DownloadTestCaseForm(request.args)

        if form.validate():
            # 校验通过
            id = form.id.data # 获取请求中的id
            testcase = db.session.query(TestCaseModel).filter(TestCaseModel.id == id).first() # 根据id在数据库中查询用例数据

            if testcase is not None:
                file_path = testcase.file_path
                file_name = file_path.split('/')[-1] # 获取用例文件名

                # 返回用例文件。使用send_from_directory来返回本地的文件，第一个参数为目录，第二个参数为文件名
                # 注意：真正的项目中我们不会将文件储存在本地目录下，而是储存在git或者其他储存服务器中
                file = send_from_directory(f'./case_files',filename=file_name,as_attachment=True)
                return file
            else:
                res = {
                    'code': 404,
                    'message': '未找到对应的用例'
                }
                return jsonify(res)以上是关于python自动化自动化测试平台开发：5.后端开发之用例文件下载，执行用例，查询用例执行状态的主要内容，如果未能解决你的问题，请参考以下文章
 
 python自动化自动化测试平台开发：6.后端开发之用例报告的上传
 python自动化自动化测试平台开发：2.flask技术讲解上
 python自动化自动化测试平台开发：3.flask技术讲解上
 自动化测试平台：前期准备和后端服务搭建
 python Django接口自动化测试
 python自动化自动化测试平台开发：1.自动化测试平台简介