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插件
def create_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 = None
try:
# 接收前端请求数据，并使用json.loads 把数据转换成Python可以读格式
resq_obj = request.data
resq_json = None
if type(resq_obj) is bytes:
resq_json = resq_obj.decode()
else:
resq_json = resq_obj
user_obj = jsonpickle.loads(resq_json) # 此处使用了jsonpickle的lib，使用json lib也是一样的
logging.debug('create user {0} '.format(user_obj))
if 'default_project_id' in user_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'\]
except Exception as ee:
abort(400)
#raise ee
if domain\_id is not None and user\_name is not None and user\_passwd is not None:
# 笔者使用 keystone 进行用户管理 此处在keystone创建帐号，然后获取id,把id及其它信息会存放在自定义的数据库表中
resp = usermgt.create\_user(domain\_id, user\_name, user\_passwd, project\_id) # resp返回值是 dict 格式
if 'id' in resp\['user'\]:
try:
# 把前端POST的表单值进行变量赋值 传给后台进行存库
user\_id = resp\['user'\]\['id'\]
if 'realname' in user\_obj\['user'\]:
real\_name = user\_obj\['user'\]\['realname'\]
if 'email' in user\_obj\['user'\]:
email = user\_obj\['user'\]\['email'\]
if 'tel' in user\_obj\['user'\]:
tel = user\_obj\['user'\]\['tel'\]
if 'phone' in user\_obj\['user'\]:
phone = user\_obj\['user'\]\['phone'\]
if 'org\_id' in user\_obj\['user'\]:
org\_id = user\_obj\['user'\]\['org\_id'\]
if 'org\_array' in user\_obj\['user'\]:
org\_array = user\_obj\['user'\]\['org\_array'\]
if 'comments' in user\_obj\['user'\]:
comments = user\_obj\['user'\]\['comments'\]
except Exception as e:
# raise e
abort(400)
if user\_obj is not None:
try:
# 调用方法进行后台存库
GuardianModelDao.\_save\_update\_users\_info(user\_id, user\_name, real\_name, email, tel, phone, org\_id, org\_array, comments)
return wrap\_response(SUCCESS\_MSG, 200) # 执行成功 返回操作成功信息及状态码 200
except Exception as e:
LOG.debug(e)
abort(500) # 否则 后台500 错误
else:
abort(500)后台函数
用于实现对接口中传递数据进行处理入库、修改、查询等等。
@classmethod
def _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:
with TxConn() as conn: # 连接数据库
with conn.cursor() as cursor:
# save user data
now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
# 接收变量赋值 user_data
users_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()
except Exception as e:
LOG.exception('error saving users info', e)
raise e
测试脚本（新增用户）
python unittest做单元测试
import unittest, requests
class GuardianTestCase(unittest.TestCase):
guardian_base = 'http://127.0.0.1:7070'
user_token = keystone.user_token('default', 'tsbc', '') # 调用keystone生成登录用户的token
def test_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": "",
"realname": "Ray",
"email": "tsbc@123.com",
"tel":"+86-029-88880000",
"phone": "",
"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: {0}'.format(resp_json))
else:
LOG.error('error create users, response: {0}, status: {1}'.format(resp_json, _resp.status_code))
接口：修改用户
修改用户和新增用户传递的参数差不多，使用的是http的patch请求，route要添加user_id的参数 ，其他实现和新增差不多代码不重复贴了
@app.route('/guardian/users/
@require_role('guardian_admin')
@login_required
def update_user(user_id):
"""
此处省略接收请求数据代码
"""
try:
resp = usermgt.update_user(domain_id, user_id, username, user_passwd, project_id, enabled) # 更新keystone 用户密码、状态
if resp is not None:
# 更新用户基本信息与新增是同一个接口，执行SQL时使用PostGresql特性，相同user_id数据只更新不会再插入
GuardianModelDao._save_update_users_info(user_id, username, real_name, email, tel, phone, org_id, org_array, comments)
return wrap_response(SUCCESS_MSG, 200) # 更新成功返回 200
except Exception as e:
LOG.exception('error modify user', e)
abort(500) # 更新失败500
测试脚本(修改用户)
def test_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": "",
"realname": "xioohua",
"email": "haoydun@123.com",
"tel": "+86-029-88880000",
"phone": "",
"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: {0}'.format(resp\_json))
else:
LOG.error('error update users, response: {0}, status: {1}'.format(resp\_json, \_resp.status\_code))接口：查询用户
根据用户名查询用户，使用http的get请求
@app.route('/guardian/users/
@require_role('guardian_admin')
@login_required
def query_user_by_username(user_name):
try:
user\_info = GuardianModelDao.\_query\_users\_info\_by\_name(user\_name) # 调用后台查询函数
if len(user\_info):
return wrap\_response(jsonpickle.dumps(user\_info), 200) # 成功返回200 并把返回内容转换成 json格式
else:
return wrap\_response("\[\]", 200) # 否则返回空
except Exception as e:
LOG.exception('error query user by name', e)
abort(500)后台函数
@classmethod
def _query_users_info_by_id(cls, user_id):
LOG.debug('start list users info…')
users = []
with TxConn() as conn:
try:
with conn.cursor() as cursor:
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) in cursor:
user = Users(user_id, real_name, email, tel, phone, org_id, org_array, comments, create_time, update_time)
users.append(user)
except Exception as e:
LOG.exception('error query users !')
raise e
return users
测试脚本(查询用户)
def test_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: {0}'.format(resp\_json))
else:
LOG.error('error query users, response: {0}, status: {1}'.format(resp\_json, \_resp.status\_code))接口：删除用户
根据用户ID进行删除，使用http的delete请求
@app.route('/guardian/users/
@require_role('guardian_admin')
@login_required
def delete_user(user_id):
if user_id is None:
abort(400)
try:
s_code = usermgt.delete_user(user_id)
if s_code == 204:
return wrap_response(SUCCESS_MSG, status=s_code) # 返回状态码
except Exception as e:
LOG.exception('error delete users info', e)
abort(500)
后台函数
def delete_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 {0}'.format(\_resp))
return \_resp.status\_code
else:
LOG.error('error list users, response: {0}'.format(\_resp))
raise KeystoneException('error delete user {0} with response status {1}'
.format(user\_id, \_resp.status\_code), \_resp.status\_code)测试脚本(删除用户)
def test_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: {0}'.format(resp\_json))
else:
LOG.error('error delete users api, response: {0}, status: {1}'.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": "",
"realname": "Ray",
"email": "haoydun@123.com",
"tel":"+86-029-88880000",
"phone": "",
"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": "",
"realname": "Ray",
"email": "tsbc@vip.qq.com",
"tel":"+86-029-88880000",
"phone": "",
"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
手机扫一扫
移动阅读更方便
你可能感兴趣的文章