专业做蛋糕的网站,新闻最新消息10条,哪类小网站容易做,建筑人才网档案查询片言
此文中代码都是笔者工作中源码#xff0c;所以不会很完整#xff0c;主要摘常见场景的api片段用以举例说明
另:此文主要针对自动化测试人员#xff0c;尤其有python基础阅读更佳。
笔者使用 python3.6 postgresql10 flask 0.12 的环境
场景介绍
我们要实现后台…片言
此文中代码都是笔者工作中源码所以不会很完整主要摘常见场景的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: tsbcvip.qq.com,tel: 86-029-88880000,phone: 13800138000,org_id: 7d22d25a-b648-4ad3-a834-2398eb989a9e,org_array:[],comments: 备注信息}}try:domain_id Noneuser_name Noneuser_passwd Noneproject_id Nonereal_name Noneemail Nonephone Noneorg_id Noneorg_array Nonecomments Nonetry:# 接收前端请求数据并使用json.loads 把数据转换成Python可以读格式resq_obj request.dataresq_json Noneif type(resq_obj) is bytes:resq_json resq_obj.decode()else:resq_json resq_objuser_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 eeif 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 eabort(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) # 执行成功 返回操作成功信息及状态码 200except Exception as e: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:with TxConn() as conn: # 连接数据库with conn.cursor() as cursor:# 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()except Exception as e:LOG.exception(error saving users info, e)raise e 测试脚本新增用户
python unittest做单元测试 import unittest, requestsclass GuardianTestCase(unittest.TestCase):guardian_base http://127.0.0.1:7070user_token keystone.user_token(default, tsbc, 123456) # 调用keystone生成登录用户的tokendef 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: 123456,realname: Ray,email: tsbc123.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, jsonuser, headersmyheaders, verifyTrue)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/string:user_id, methods[PATCH])
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) # 更新成功返回 200except 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_tokenuser {user: {domain_id: default,enabled: True,name: tsbc1,password: 111111,realname: xioohua,email: haoydun123.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方法jsonuser, headersmyheaders, verifyTrue)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/string:user_name, methods[GET]) # 查询使用 GET 方法
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) 后台函数 classmethoddef _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 ereturn 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, headersmyheaders, verifyTrue) # 链接中加入用户名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/string:user_id, methods[DELETE]) # 传入用户id 使用 delete请求
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, statuss_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,headersmyheaders, verifyFalse)if _resp.status_code 204:LOG.debug(delete users -- response {0}.format(_resp))return _resp.status_codeelse: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, headersmyheaders, # 使用 delete 请求verifyTrue)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: 111111,realname: Ray,email: haoydun123.com,tel:86-029-88880000,phone: 13800138000,org_id: 7d22d25a-b648-4ad3-a834-2398eb989a9e,org_array: null,comments: 备注信息}
} ## UPDATE USER
1. URL: /guardian/users/string:user_id
2. METHOD: PATCH2. body: {user: {domain_id: default,enabled: True,name: tsbc1,password: 111111,realname: Ray,email: tsbcvip.qq.com,tel:86-029-88880000,phone: 13800138000,org_id: 7d22d25a-b648-4ad3-a834-2398eb989a9e,org_array: null,comments: 备注信息}} 该接口可以用于修改密码禁用用户
## QUERY USER BY NAME
1. URL: /guardian/users/string:user_name
2. METHOD: GET
## DELETE USER
1. URL: /guardian/users/string:user_id
2. METHOD: DELETE 2023最新Jmeter接口测试从入门到精通全套项目实战教程
