Flask 编写http接口api及接口自动化测试

片言

此文中代码都是笔者工作中源码，所以不会很完整，主要摘常见场景的api片段用以举例说明

另:此文主要针对自动化测试人员，尤其有python基础阅读更佳。

笔者使用 python3.6 + postgresql10 + flask 0.12 的环境

场景介绍

我们要实现后台管理中的用户管理功能，实现对用户的增、修改、删除、查询操作。

文中场景省去登录和用户权限部分，但我会在接口代码片段中进行注释说明

接口：新增用户

（定义访问的http链接地址，以及请求使用的 method）

http://127.0.0.1:7070/guardian/users

@app.route('/guardian/users',methods=['POST'])# 使用 POST 方法

@require_role('guardian_admin')# 登录用户的角色权限必须是guardian_admin 此处是自定义的更角色管理中的设置管理员角色一致即可。不需要角色权限也可以注释此行即可

@login_required# 验证用户登录 flask_login插件

defcreate_user():''' 下面json串是新增时候前台表单提交的内容以及格式，也就是我们接口测试时候需要添加的测试数据 { "user": { "domain_id": "default", "enabled": True, "name": "tsbc1", "password": "111111", "realname": "Ray", "email": "tsbc@vip.qq.com", "tel": "+86-029-88880000", "phone": "13800138000", "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e", "org_array":[], "comments": "备注信息" } } '''try:domain_id=None user_name=None user_passwd=None project_id=None real_name=None email=None phone=None org_id=None org_array=None comments=Nonetry:# 接收前端请求数据，并使用json.loads 把数据转换成Python可以读格式resq_obj=request.data resq_json=Noneiftype(resq_obj)isbytes:resq_json=resq_obj.decode()else:resq_json=resq_obj user_obj=jsonpickle.loads(resq_json)# 此处使用了jsonpickle的lib，使用json lib也是一样的logging.debug('create user '.format(user_obj))if'default_project_id'inuser_obj['user']:project_id=user_obj['user']['default_project_id']domain_id=user_obj['user']['domain_id']user_name=user_obj['user']['name']user_passwd=user_obj['user']['password']exceptExceptionasee:abort(400)#raise eeifdomain_idisnotNoneanduser_nameisnotNoneanduser_passwdisnotNone:# 笔者使用 keystone 进行用户管理 此处在keystone创建帐号，然后获取id,把id及其它信息会存放在自定义的数据库表中resp=usermgt.create_user(domain_id,user_name,user_passwd,project_id)# resp返回值是 dict 格式if'id'inresp['user']:try:# 把前端POST的表单值进行变量赋值 传给后台进行存库user_id=resp['user']['id']if'realname'inuser_obj['user']:real_name=user_obj['user']['realname']if'email'inuser_obj['user']:email=user_obj['user']['email']if'tel'inuser_obj['user']:tel=user_obj['user']['tel']if'phone'inuser_obj['user']:phone=user_obj['user']['phone']if'org_id'inuser_obj['user']:org_id=user_obj['user']['org_id']if'org_array'inuser_obj['user']:org_array=user_obj['user']['org_array']if'comments'inuser_obj['user']:comments=user_obj['user']['comments']exceptExceptionase:# raise eabort(400)ifuser_objisnotNone:try:# 调用方法进行后台存库GuardianModelDao._save_update_users_info(user_id,user_name,real_name,email,tel,phone,org_id,org_array,comments)returnwrap_response(SUCCESS_MSG,200)# 执行成功 返回操作成功信息及状态码 200exceptExceptionase:LOG.debug(e)abort(500)# 否则 后台500 错误else:abort(500)

后台函数

用于实现对接口中传递数据进行处理入库、修改、查询等等。

@classmethoddef_save_update_users_info(cls,user_id,username,real_name,email,tel,phone,org_id,org_array,comments):LOG.debug('start create users info...')try:withTxConn()asconn:# 连接数据库withconn.cursor()ascursor:# save user datanow_time=time.strftime("%Y-%m-%d %H:%M:%S",time.localtime())# 接收变量赋值 user_datausers_data={"user_id":user_id,"username":username,"realname":real_name,"email":email,"tel":tel,"phone":phone,"org":org_id,"org_array":org_array,"comments":comments,"now_time":now_time}cursor.execute(cls.insert_users_info,users_data)# 插入数据conn.commit()exceptExceptionase:LOG.exception('error saving users info',e)raisee

测试脚本（新增用户）

python unittest做单元测试

importunittest,requests

classGuardianTestCase(unittest.TestCase):guardian_base='http://127.0.0.1:7070'user_token=keystone.user_token('default','tsbc','123456')# 调用keystone生成登录用户的tokendeftest_create_user_api(self):# 添加对应headers 及 tonken 用于数据传参和登录认证使用myheaders={'Content-Type':'application/json'}myheaders['X-Subject-Token']=self.user_token# 接口测试数据user={"user":{"domain_id":"default","enabled":True,"name":"tsbc1","password":"123456","realname":"Ray","email":"tsbc@123.com","tel":"+86-029-88880000","phone":"13800138000","org_id":"7d22d25a-b648-4ad3-a834-2398eb989a9e","org_array":None,"comments":"备注信息"}}# 使用requests的post方法进行请求路由_resp=requests.post(self.guardian_base+'/guardian/users',json=user,headers=myheaders,verify=True)resp_json=_resp.content# 对返回结果及状态进行校验if_resp.ok:self.assertIsNotNone(resp_json)LOG.info('create users, response: '.format(resp_json))else:LOG.error('error create users, response: , status: '.format(resp_json,_resp.status_code))

接口：修改用户

修改用户和新增用户传递的参数差不多，使用的是http的patch请求，route要添加user_id的参数 ，其他实现和新增差不多代码不重复贴了

@app.route('/guardian/users/',methods=['PATCH'])@require_role('guardian_admin')@login_requireddefupdate_user(user_id):""" 此处省略接收请求数据代码 """try:resp=usermgt.update_user(domain_id,user_id,username,user_passwd,project_id,enabled)# 更新keystone 用户密码、状态ifrespisnotNone:# 更新用户基本信息与新增是同一个接口，执行SQL时使用PostGresql特性，相同user_id数据只更新不会再插入GuardianModelDao._save_update_users_info(user_id,username,real_name,email,tel,phone,org_id,org_array,comments)returnwrap_response(SUCCESS_MSG,200)# 更新成功返回 200exceptExceptionase:LOG.exception('error modify user',e)abort(500)# 更新失败500

测试脚本(修改用户)

deftest_update_user_api(self):myheaders={'Content-Type':'application/json'}myheaders['X-Subject-Token']=self.user_token user={"user":{"domain_id":"default","enabled":True,"name":"tsbc1","password":"111111","realname":"xioohua","email":"haoydun@123.com","tel":"+86-029-88880000","phone":"13800138000","org_id":"7d22d25a-b648-4ad3-a834-2398eb989a9e","comments":"备注信息"}}_resp=requests.patch(self.guardian_base+'/guardian/users/0d26176579e94d558c4c2d864b003241',# 路由中添加要修改的用户ID requests使用 patch方法json=user,headers=myheaders,verify=True)resp_json=_resp.content# 返回值进行校验if_resp.ok:self.assertIsNotNone(resp_json)LOG.info('update users, response: '.format(resp_json))else:LOG.error('error update users, response: , status: '.format(resp_json,_resp.status_code))

接口：查询用户

根据用户名查询用户，使用http的get请求

@app.route('/guardian/users/',methods=['GET'])# 查询使用 GET 方法@require_role('guardian_admin')@login_requireddefquery_user_by_username(user_name):try:user_info=GuardianModelDao._query_users_info_by_name(user_name)# 调用后台查询函数iflen(user_info):returnwrap_response(jsonpickle.dumps(user_info),200)# 成功返回200 并把返回内容转换成 json格式else:returnwrap_response("[]",200)# 否则返回空exceptExceptionase:LOG.exception('error query user by name',e)abort(500)

后台函数

@classmethoddef_query_users_info_by_id(cls,user_id):LOG.debug('start list users info...')users=[]withTxConn()asconn:try:withconn.cursor()ascursor:data=(user_id,)cursor.execute(cls.query_users_info_by_id,data)# 查询结构如果是行数据，放进list中进行返回for(user_id,real_name,email,tel,phone,org_id,org_array,comments,create_time,update_time)incursor:user=Users(user_id,real_name,email,tel,phone,org_id,org_array,comments,create_time,update_time)users.append(user)exceptExceptionase:LOG.exception('error query users !')raiseereturnusers

测试脚本(查询用户)

deftest_query_user_api(self):myheaders={'Content-Type':'application/json'}myheaders['X-Subject-Token']=self.user_token _resp=requests.get(self.guardian_base+'/guardian/users/tsbc',headers=myheaders,verify=True)# 链接中加入用户名resp_json=_resp.text# 验证判断if_resp.ok:self.assertIsNotNone(resp_json)LOG.info('query users, response: '.format(resp_json))else:LOG.error('error query users, response: , status: '.format(resp_json,_resp.status_code))

接口：删除用户

根据用户ID进行删除，使用http的delete请求

@app.route('/guardian/users/',methods=['DELETE'])# 传入用户id 使用 delete请求@require_role('guardian_admin')@login_requireddefdelete_user(user_id):ifuser_idisNone:abort(400)try:s_code=usermgt.delete_user(user_id)ifs_code==204:returnwrap_response(SUCCESS_MSG,status=s_code)# 返回状态码exceptExceptionase:LOG.exception('error delete users info',e)abort(500)

后台函数

defdelete_user(user_id):myheaders={'Content-Type':'application/json'}myheaders['X-Auth-Token']=settings.KEYSTONE_ADMIN_TOKEN _resp=requests.delete(settings.KEYSTONE_BASE+'/v3/users/'+user_id,headers=myheaders,verify=False)if_resp.status_code==204:LOG.debug('delete users -- response '.format(_resp))return_resp.status_codeelse:LOG.error('error list users, response: '.format(_resp))raiseKeystoneException('error delete user with response status '.format(user_id,_resp.status_code),_resp.status_code)

测试脚本(删除用户)

deftest_delete_users_api(self):myheaders={'Content-Type':'application/json'}myheaders['X-Subject-Token']=self.user_token _resp=requests.delete(self.guardian_base+'/guardian/users/tsbc1',headers=myheaders,# 使用 delete 请求verify=True)resp_json=_resp.content# 根据返回状态进行校验if_resp.ok:self.assertIsNotNone(resp_json)LOG.info('delete users api, response: '.format(resp_json))else:LOG.error('error delete users api, response: , status: '.format(resp_json,_resp.status_code))

下面提供给前端API文档

## CREATE USER

1. URL:' /guardian/users'

2. METHOD: POST

3. body:

{

"user": {

"domain_id": "default",

"enabled": True,

"name": "tsbc1",

"password": "111111",

"realname": "Ray",

"org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",

"org_array": null,

"comments": "备注信息"

}

}

## UPDATE USER

1. URL:' /guardian/users/'

2. METHOD: PATCH

2. body:

{

"user": {

"domain_id": "default",

"enabled": True,

"name": "tsbc1",

"password": "111111",

"realname": "Ray",

"org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",

"org_array": null,

"comments": "备注信息"

}

}

该接口可以用于修改密码，禁用用户

## QUERY USER BY NAME

1. URL:' /guardian/users/'

2. METHOD: GET

## DELETE USER

1. URL: '/guardian/users/'

2. METHOD: DELETE

作者： Ray

博客地址：http://www.cnblogs.com/tsbc/

声明：本文已获作者授权转载，著作权归作者所有。