背景: 1.平时测试接口,总是现写代码,对测试用例的管理,以及测试报告的管理持久化做的不够,
2.工作中移动端开发和后端开发总是不能并行进行,需要一个 mock 的依赖来让他们并行开发。
3.同时让自己锻炼去开发测试平台,掌握 flask 开发程序,提高自己的业务水平。
整体思路: 1.利用 flask+bootstrap 来进行 web 界面开发,对接口,接口测试用例,定时任务,测试报告的持续集成。
2.IAPTest 支持接口用例管理,接口多用例测试,支持定时测试任务,测试报告持久化
3.目前 mock 服务支持单一 path,定时任务可以开启暂停多用例执行,定时任务执行后自动发送测试报告,多用例的单次执行,单接口的调试功能。对测试环境的管理
下面来看下最后的效果图,以及附上 github 开源地址。
测试环境管理界面:
定时任务界面:
mock 界面
测试报告界面
用例管理界面
接口管理界面
核心代码分享区:
定时任务对应视图开发
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 代码
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 服务的一个请求方式的代码
class MakemockserverView(MethodView):#做一个mock服务
def get(self,path):#get请求方法
huoqupath=Mockserver.query.filter_by(path=path,status=True).first()
heders=request.headers
method=request.method
if not huoqupath:
abort(404)
if method.lower() !=huoqupath.methods:
return jsonify({'code':'-1','message':'请求方式错误!','data':''})
if huoqupath.is_headers==True:
if comp_dict(heders,huoqupath.headers) ==True:
if huoqupath.ischeck==True:
paerm = request.values.to_dict()
if dict_par(paerm,huoqupath.params)==True:
if huoqupath.rebacktype == 'json':
try:
json_fan = json.dumps(huoqupath.fanhui)
return jsonify({'code': '1', 'message': 'successs', 'data': json_fan})
except:
return jsonify({'code': '-2', 'message': '你写入的返回不能正常json!请检查', 'data': ''})
elif huoqupath.rebacktype == 'xml':
response = make_response(huoqupath.fanhui)
response.content_type = 'application/xml'
return response
else:
return jsonify({'code': '-2', 'message': '你写入的类型目前系统不支持', 'data': ''})
else:
return jsonify({'code': '-4', 'message': '你输入的参数不正确', 'data': ''})
else:
if huoqupath.rebacktype=='json':
try:
json_fan=json.dumps(huoqupath.fanhui)
return jsonify({'code': '1', 'message': 'successs', 'data':json_fan})
except:
return jsonify({'code': '-2', 'message': '你写入的返回不能正常json!请检查', 'data': ''})
elif huoqupath.rebacktype =='xml':
response=make_response(huoqupath.fanhui)
response.content_type='application/xml'
return response
else:
return jsonify({'code': '-2', 'message': '你写入的类型目前系统不支持', 'data': ''})
else:
return jsonify({'code': '-3', 'message': '安全校验失败!', 'data': ''})
if huoqupath.ischeck == True:
paerm = request.values.to_dict()
if dict_par(paerm, huoqupath.params) == True:
if huoqupath.rebacktype == 'json':
try:
json_fan = json.dumps(huoqupath.fanhui)
return jsonify({'code': '1', 'message': 'successs', 'data': json_fan})
except:
return jsonify({'code': '-2', 'message': '你写入的返回不能正常json!请检查', 'data': ''})
elif huoqupath.rebacktype == 'xml':
response = make_response(huoqupath.fanhui)
response.content_type = 'application/xml'
return response
else:
return jsonify({'code': '-2', 'message': '你写入的类型目前系统不支持', 'data': ''})
else:
return jsonify({'code': '-4', 'message': '你输入的参数不正确', 'data': ''})
else:
if huoqupath.rebacktype == 'json':
try:
json_fan = json.dumps(huoqupath.fanhui)
return jsonify({'code': '1', 'message': 'successs', 'data': json_fan})
except:
return jsonify({'code': '-2', 'message': '你写入的返回不能正常json!请检查', 'data': ''})
elif huoqupath.rebacktype == 'xml':
response = make_response(huoqupath.fanhui)
response.content_type = 'application/xml'
return response
else:
return jsonify({'code': '-2', 'message': '你写入的类型目前系统不支持', 'data': ''}) #
1.依赖包为 requirements.txt 文件下的,可能部分不全,需要什么可以自己使用中增加。
2.目前由于考虑后续迁移内部使用的话,所以移除了注册功能,改为管理员后台添加方式,默认登录:liwanlei 密码:liwanlei
3.部分功能调试还存在一定的问题,欢迎各位多提宝贵意见,
大家可以多提意见,后续会优化,最近一直熬夜加班可能有些消息不能及时回复,还望谅解。
项目开发中,有些功能的开发参照了论坛部分作者的文章给我学习上面指引了方向,感恩 testerhome,