flask + Python3 實現的的API自動化測試平臺---- IAPTest接口測試平臺，改名：FXTest 接受定製開發（java版開發完畢）

時間 2019-11-20

標籤 flask python3 python 實現 api 自動化測試平臺 iaptest 接口改名 fxtest 接受定製開發 java 完畢欄目 Python 简体版

原文原文鏈接

**背景： 1.平時測試接口，老是現寫代碼，對測試用例的管理，以及測試報告的管理持久化作的不夠，
              2.工做中移動端開發和後端開發老是不能並行進行，須要一個mock的依賴來讓他們並行開發。
              3.同時讓本身鍛鍊去開發測試平臺，掌握flask開發程序，提升本身的業務水平。

總體思路：    1.利用flask+bootstrap來進行web界面開發，對接口，接口測試用例，定時任務，測試報告的持續集成。
                           2.IAPTest支持接口用例管理，接口多用例測試，支持定時測試任務，測試報告持久化
                           3.目前mock服務支持單一path，定時任務能夠開啓暫停多用例執行，定時任務執行後自動發送測試報告，多用例的單次執行，單接口的調試功能。對測試環境的管理
下面來看下最後的效果圖，以及附上github開源地址。

測試環境管理界面：

html

定時任務界面：

git

mock界面

github

測試報告界面

web

用例管理界面
ajax

接口管理界面

json

**核心代碼分享區：**

定時任務對應視圖開發

flask

class AddtimingtaskView(MethodView):
    @login_required
    def get(self):
        return  render_template('addtimingtasks.html')
    @login_required
    def post(self):
        taskname=request.form['taskname']
        tinmingtime=request.form['time']
        to_email_data=request.form['to_email']
        cao_email=request.form['cao_email']
        weihu=request.form['weihu']
        if taskname =='':
            flash('任務名不能爲空！')
            return render_template('addtimingtasks.html')
        if tinmingtime =='':
            flash('任務執行時間不能爲空！')
            return render_template('addtimingtasks.html')
        if to_email_data=='':
            flash('發送給誰郵件不能爲空！')
            return render_template('addtimingtasks.html')
        if weihu=='':
            flash('維護人郵件不能爲空！')
            return render_template('addtimingtasks.html')
        taskname_is = Task.query.filter_by(taskname=taskname).first()
        if taskname_is:
            flash('任務已經存在請從新填寫！')
            return render_template('addtimingtasks.html')
        new_task=Task(taskname=taskname,taskstart=tinmingtime,taskrepor_to=to_email_data,taskrepor_cao=cao_email,task_make_email=weihu,
                      makeuser=current_user.id)
        db.session.add(new_task)
        try:
            db.session.commit()
            flash('添加定時任務成功')
            return  redirect(url_for('timingtask'))
        except Exception as e:
            db.session.rollback()
            flash('添加過程貌似異常艱難！')
            return redirect(url_for('addtimingtasks'))
        return render_template('addtimingtasks.html')
class Editmingtaskview(MethodView):
    @login_required
    def get(self,id):
        task_one=Task.query.filter_by(id=id).first()
        procjet=Project.query.all()
        if not task_one:
            flash('你編輯的不存在')
            return  redirect(url_for('timingtask'))
        return  render_template('Edittimingtasks.html',task_one=task_one,porjects=procjet)
    def post(self,id):
        task_one = Task.query.filter_by(id=id).first()
        procjet = Project.query.all()
        taskname = request.form['taskname']
        tinmingtime = request.form['time']
        to_email_data = request.form['to_email']
        cao_email = request.form['cao_email']
        weihu = request.form['weihu']
        if taskname =='':
            flash('任務名不能爲空！')
            return render_template('addtimingtasks.html')
        if tinmingtime =='':
            flash('任務執行時間不能爲空！')
            return render_template('addtimingtasks.html')
        if to_email_data=='':
            flash('發送給誰郵件不能爲空！')
            return render_template('addtimingtasks.html')
        if weihu=='':
            flash('維護人郵件不能爲空！')
            return render_template('addtimingtasks.html')
        task_one.taskname=taskname
        task_one.taskrepor_to=to_email_data
        task_one.taskrepor_cao=cao_email
        task_one.task_make_email=weihu
        task_one.makeuser=current_user.id
        try:
            db.session.commit()
            flash('編輯成功')
            return  redirect(url_for('timingtask'))
        except:
            db.session.rollback()
            flash('編輯出現問題！')
            return redirect(url_for('timingtask'))
        return render_template('Edittimingtasks.html', task_one=task_one,porjects=procjet)
class DeteleTaskViee(MethodView):
    def get(self,id):
        task_one = Task.query.filter_by(id=id).first()
        if not task_one:
            flash('你編輯的不存在')
            return redirect(url_for('timingtask'))
        if task_one.status==True:
            flash('已經刪除')
            return redirect(url_for('timingtask'))
        task_one.status=True
        try:
            db.session.commit()
            flash('刪除任務成功')
            return redirect(url_for('timingtask'))
        except:
            db.session.rollback()
            flash('刪除任務休息了')
            return redirect(url_for('timingtask'))
@app.route('/gettest',methods=['POST'])
@login_required
def gettest():#ajax獲取項目的測試用例
    projec=(request.get_data('project')).decode('utf-8')
    if not projec:
        return []
    proje=Project.query.filter_by(project_name=str(projec)).first()
    if not proje:
        return  []
    testyong=InterfaceTest.query.filter_by(projects_id=proje.id).all()
    testyong_list=[]
    for i in testyong:
        testyong_list.append({'name':i.Interface_name,'id':i.id})
    return   jsonify({'data':testyong_list})
class TestforTaskView(MethodView):#爲測試任務添加測試用例
    def get(self,id):
        procjet = Project.query.all()
        task_one=Task.query.filter_by(id=id).first()
        return  render_template('addtestyongfortask.html',task_one=task_one,procjets=procjet)
    def post(self,id):
        procjet = Project.query.all()
        task_one = Task.query.filter_by(id=id).first()
        proc_test=request.form.get('project')
        if proc_test =='':
            flash(u'不能不添加測試項目！')
            return render_template('addtestyongfortask.html', task_one=task_one, procjets=procjet)
        test_yongli=request.form.getlist('testyongli')
        if test_yongli=='':
            flash(u'親你見過只有測試項目沒有測試用例的測試任務嗎！')
            return render_template('addtestyongfortask.html', task_one=task_one, procjets=procjet)
        for oldtask in task_one.interface.all():
            task_one.interface.remove(oldtask)
        task_one.prject=Project.query.filter_by(project_name=proc_test).first().id
        for yongli in test_yongli:
            task_one.interface.append(InterfaceTest.query.filter_by(id=yongli).first())
            db.session.add(task_one)
        try:
            db.session.commit()
            flash('任務更新用例成功')
            return  redirect(url_for('timingtask'))
        except:
            flash('任務更新用例失敗')
            return redirect(url_for('timingtask'))
        return render_template('addtestyongfortask.html', task_one=task_one, procjets=procjet)
class StartTaskView(MethodView):#開始定時任務
    @login_required
    def get(self,id):
        task=Task.query.filter_by(id=id).first()
        if len(task.interface.all())<=1:
            flash('定時任務執行過程的測試用例爲多用例，請你諒解')
            return  redirect(url_for('timingtask'))
        try:
            scheduler.add_job(func=addtask, id=str(id), args=str(id),trigger=eval(task.taskstart),replace_existing=True)
            task.yunxing_status='啓動'
            db.session.commit()
            flash(u'定時任務啓動成功！')
            return  redirect(url_for('timingtask'))
        except Exception as e:
            flash(u'定時任務啓動失敗！請檢查任務的各項內容各項內容是否正常')
            return redirect(url_for('timingtask'))
class ZantingtaskView(MethodView):#暫停定時任務
    @login_required
    def get(self,id):
        task = Task.query.filter_by(id=id).first()
        try:
            scheduler.pause_job(str(id))
            task.yunxing_status = '暫停'
            db.session.commit()
            flash(u'定時任務暫停成功！')
            return redirect(url_for('timingtask'))
        except:
            task.yunxing_status = '建立'
            db.session.commit()
            flash(u'定時任務暫停失敗！已經爲您初始化')
            return redirect(url_for('timingtask'))
class HuifutaskView(MethodView):#回覆定時任務
    @login_required
    def get(self,id):
        task = Task.query.filter_by(id=id).first()
        try:
            scheduler.resume_job(str(id))
            task.yunxing_status='啓動'
            db.session.commit()
            flash(u'定時任務恢復成功！')
            return redirect(url_for('timingtask'))
        except:
            task.yunxing_status = '建立'
            db.session.commit()
            flash(u'定時任務恢復失敗！已經爲您初始化')
            return redirect(url_for('timingtask'))
class YichuTaskView(MethodView):#移除定時任務
    @login_required
    def get(self,id):
        task = Task.query.filter_by(id=id).first()
        try:
            scheduler.delete_job(str(id))
            task.yunxing_status='關閉'
            db.session.commit()
            flash(u'定時任務移除成功！')
            return redirect(url_for('timingtask'))
        except:
            task.yunxing_status = '建立'
            db.session.commit()
            flash(u'定時任務移除失敗！已經爲您初始化')
            return redirect(url_for('timingtask'))

定時任務所執行的func代碼bootstrap

def addtask(id):#定時任務執行的時候所用的函數
    in_id=int(id)
    task=Task.query.filter_by(id=in_id).first()
    starttime = datetime.datetime.now()
    star = time.time()
    day = time.strftime("%Y%m%d%H%M", time.localtime(time.time()))
    basedir = os.path.abspath(os.path.dirname(__file__))
    file_dir = os.path.join(basedir, 'upload')
    file = os.path.join(file_dir, (day + '.log'))
    if os.path.exists(file) is False:
        os.system('touch %s' % file)
    filepath = os.path.join(file_dir, (day + '.html'))
    if os.path.exists(filepath) is False:
        os.system(r'touch %s' % filepath)
    projecct_list = []
    model_list = []
    Interface_name_list = []
    Interface_url_list = []
    Interface_meth_list = []
    Interface_pase_list = []
    Interface_assert_list = []
    Interface_headers_list = []
    id_list = []
    for task_yongli in task.interface.all():
        id_list.append(task_yongli.id)
        projecct_list.append(task_yongli.projects)
        model_list.append(task_yongli.models)
        Interface_url_list.append(task_yongli.Interface_url)
        Interface_name_list.append(task_yongli.Interface_name)
        Interface_meth_list.append(task_yongli.Interface_meth)
        Interface_pase_list.append(task_yongli.Interface_pase)
        Interface_assert_list.append(task_yongli.Interface_assert)
        Interface_headers_list.append(task_yongli.Interface_headers)
    apitest = ApiTestCase(Interface_url_list, Interface_meth_list, Interface_pase_list, Interface_assert_list, file,
                          Interface_headers_list)
    result_toal, result_pass, result_fail, relusts, bask_list = apitest.testapi()
    endtime = datetime.datetime.now()
    end = time.time()
    createHtml(titles=u'接口測試報告', filepath=filepath, starttime=starttime, endtime=endtime, passge=result_pass,
               fail=result_fail, id=id_list, name=projecct_list, headers=Interface_headers_list,
               coneent=Interface_url_list, url=Interface_meth_list, meth=Interface_pase_list,
               yuqi=Interface_assert_list, json=bask_list, relusts=relusts)
    hour = end - star
    user_id = User.query.filter_by(role_id=2).first().id
    new_reust = TestResult(Test_user_id=user_id, test_num=result_toal, pass_num=result_pass, fail_num=result_fail,
                           test_time=starttime, hour_time=hour, test_rep=(day + '.html'), test_log=(day + '.log'))
    email = EmailReport.query.filter_by(role_id=2, default_set=True).first()
    send_emails(sender=email.send_email, receivers=task.taskrepor_to, password=email.send_email_password,
                smtp=email.stmp_email, port=email.port, fujian1=file, fujian2=filepath, subject=u'%s自動用例執行測試報告' % day,
                url='%stest_rep'%(request.url_root))
    db.session.add(new_reust)
    db.session.commit()

mock服務的一個請求方式的代碼

後端

 1 class MakemockserverView(MethodView):#作一個mock服務
 2     def get(self,path):#get請求方法
 3         huoqupath=Mockserver.query.filter_by(path=path,status=True).first()
 4         heders=request.headers
 5         method=request.method
 6         if not huoqupath:
 7             abort(404)
 8         if method.lower() !=huoqupath.methods:
 9             return  jsonify({'code':'-1','message':'請求方式錯誤!','data':''})
10         if huoqupath.is_headers==True:
11             if comp_dict(heders,huoqupath.headers) ==True:
12                 if huoqupath.ischeck==True:
13                     paerm = request.values.to_dict()
14                     if dict_par(paerm,huoqupath.params)==True:
15                         if huoqupath.rebacktype == 'json':
16                             try:
17                                 json_fan = json.dumps(huoqupath.fanhui)
18                                 return jsonify({'code': '1', 'message': 'successs', 'data': json_fan})
19                             except:
20                                 return jsonify({'code': '-2', 'message': '你寫入的返回不能正常json！請檢查', 'data': ''})
21                         elif huoqupath.rebacktype == 'xml':
22                             response = make_response(huoqupath.fanhui)
23                             response.content_type = 'application/xml'
24                             return response
25                         else:
26                             return jsonify({'code': '-2', 'message': '你寫入的類型目前系統不支持', 'data': ''})
27                     else:
28                         return jsonify({'code': '-4', 'message': '你輸入的參數不正確', 'data': ''})
29                 else:
30                     if huoqupath.rebacktype=='json':
31                         try:
32                             json_fan=json.dumps(huoqupath.fanhui)
33                             return  jsonify({'code': '1', 'message': 'successs', 'data':json_fan})
34                         except:
35                             return jsonify({'code': '-2', 'message': '你寫入的返回不能正常json！請檢查', 'data': ''})
36                     elif huoqupath.rebacktype =='xml':
37                         response=make_response(huoqupath.fanhui)
38                         response.content_type='application/xml'
39                         return response
40                     else:
41                         return jsonify({'code': '-2', 'message': '你寫入的類型目前系統不支持', 'data': ''})
42             else:
43                 return jsonify({'code': '-3', 'message': '安全校驗失敗!', 'data': ''})
44         if huoqupath.ischeck == True:
45             paerm = request.values.to_dict()
46             if dict_par(paerm, huoqupath.params) == True:
47                 if huoqupath.rebacktype == 'json':
48                     try:
49                         json_fan = json.dumps(huoqupath.fanhui)
50                         return jsonify({'code': '1', 'message': 'successs', 'data': json_fan})
51                     except:
52                         return jsonify({'code': '-2', 'message': '你寫入的返回不能正常json！請檢查', 'data': ''})
53                 elif huoqupath.rebacktype == 'xml':
54                     response = make_response(huoqupath.fanhui)
55                     response.content_type = 'application/xml'
56                     return response
57                 else:
58                     return jsonify({'code': '-2', 'message': '你寫入的類型目前系統不支持', 'data': ''})
59             else:
60                 return jsonify({'code': '-4', 'message': '你輸入的參數不正確', 'data': ''})
61         else:
62             if huoqupath.rebacktype == 'json':
63                 try:
64                     json_fan = json.dumps(huoqupath.fanhui)
65                     return jsonify({'code': '1', 'message': 'successs', 'data': json_fan})
66                 except:
67                     return jsonify({'code': '-2', 'message': '你寫入的返回不能正常json！請檢查', 'data': ''})
68             elif huoqupath.rebacktype == 'xml':
69                 response = make_response(huoqupath.fanhui)
70                 response.content_type = 'application/xml'
71                 return response
72             else:
73                 return jsonify({'code': '-2', 'message': '你寫入的類型目前系統不支持', 'data': ''}) #

開源地址：https://github.com/liwanlei/FXTest

使用說明：

1.依賴包爲requirements.txt文件下的，可能部分不全，須要什麼能夠本身使用中增長。
2.目前因爲考慮後續遷移內部使用的話，因此移除了註冊功能，改成管理員後臺添加方式，默認登陸：liwanlei 密碼：liwanlei
3.部分功能調試還存在必定的問題，歡迎各位多提寶貴意見，api

部分功能欠缺：

1.定時任務的持久化，如今處理容易受到運行過程當中的宕機等狀況從新啓動服務器的定時任務所有須要開啓
二、mock接口只能支持單一的path
3. 測試環境沒有改成動態配置，動態支持。目前測試環境管理以及上線
4。部分地方可能還會有不嚴謹性，可是工具可使用。
五、目前僅支持 http請求中的json格式的，
6.你們能夠多提意見，後續會優化，最近一直熬夜加班可能有些消息不能及時回覆，還望諒解。

做者寄語：

前進的道路咱們充滿着迷茫，

前進的每一步咱們都會有收穫。

路在腳下，咱們決定不了咱們的出身，可是咱們能夠努力改變咱們將來。

告別昨天失敗的本身，努力拼搏今天，成就美好明天

有問題能夠聯繫我：QQ:952943386 email:leileili126@163.com qq羣：194704520 新羣：683894834

我的公衆號

微信打賞