devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

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

作者：笔触狂放9 | 2024-04-16 11:06:19

赞

踩

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

片言

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

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

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

场景介绍 

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

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

接口：新增用户

（定义访问的http链接地址，以及请求使用的 method）    http://127.0.0.1:7070/guardian/users

  1.  
  2. @app.route('/guardian/users', methods=['POST'])    # 使用 POST 方法
  3. @require_role('guardian_admin')    # 登录用户的角色权限必须是guardian_admin 此处是自定义的更角色管理中的设置管理员角色一致即可。不需要角色权限也可以注释此行即可
  4. @login_required    # 验证用户登录 flask_login插件
  5. def create_user():
  6.     '''
  7.     下面json串是新增时候前台表单提交的内容以及格式，也就是我们接口测试时候需要添加的测试数据
  8.     {
  9.             "user": {
  10.                 "domain_id": "default",
  11.                 "enabled": True,
  12.                 "name": "tsbc1",
  13.                 "password": "111111",
  14.                 "realname": "Ray",
  15.                 "email": "tsbc@vip.qq.com",
  16.                 "tel": "+86-029-88880000",
  17.                 "phone": "13800138000",
  18.                 "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
  19.                 "org_array":[],
  20.                 "comments": "备注信息"
  21.             }
  22.         }
  23.     '''
  24.     try:
  25.         domain_id = None
  26.         user_name = None
  27.         user_passwd = None
  28.         project_id = None
  29.         real_name = None
  30.         email = None
  31.         phone = None
  32.         org_id = None
  33.         org_array = None
  34.         comments = None
  35.         try:
  36.             # 接收前端请求数据，并使用json.loads 把数据转换成Python可以读格式
  37.             resq_obj = request.data
  38.             resq_json = None
  39.             if type(resq_obj) is bytes:
  40.                 resq_json = resq_obj.decode()
  41.             else:
  42.                 resq_json = resq_obj
  43.             user_obj = jsonpickle.loads(resq_json)    # 此处使用了jsonpickle的lib，使用json lib也是一样的
  44.             logging.debug('create user {0} '.format(user_obj))
  45.             if 'default_project_id' in user_obj['user']:
  46.                 project_id = user_obj['user']['default_project_id']
  47.  
  48.             domain_id = user_obj['user']['domain_id']
  49.             user_name = user_obj['user']['name']
  50.             user_passwd = user_obj['user']['password']
  51.  
  52.         except Exception as ee:
  53.             abort(400)
  54.             #raise ee
  55.  
  56.         if domain_id is not None and user_name is not None and user_passwd is not None:
  57.             # 笔者使用 keystone 进行用户管理 此处在keystone创建帐号，然后获取id,把id及其它信息会存放在自定义的数据库表中
  58.             resp = usermgt.create_user(domain_id, user_name, user_passwd, project_id)    # resp返回值是 dict 格式
  59.             if 'id' in resp['user']:
  60.                 try:
  61.                     # 把前端POST的表单值进行变量赋值 传给后台进行存库
  62.                     user_id = resp['user']['id']
  63.                     if 'realname' in user_obj['user']:
  64.                         real_name = user_obj['user']['realname']
  65.                     if 'email' in user_obj['user']:
  66.                         email = user_obj['user']['email']
  67.                     if 'tel' in user_obj['user']:
  68.                         tel = user_obj['user']['tel']
  69.                     if 'phone' in user_obj['user']:
  70.                         phone = user_obj['user']['phone']
  71.                     if 'org_id' in user_obj['user']:
  72.                         org_id = user_obj['user']['org_id']
  73.                     if 'org_array' in user_obj['user']:
  74.                         org_array = user_obj['user']['org_array']
  75.                     if 'comments' in user_obj['user']:
  76.                         comments = user_obj['user']['comments']
  77.  
  78.                 except Exception as e:
  79.                     # raise e
  80.                     abort(400)
  81.                 if user_obj is not None:
  82.                     try:
  83.                         # 调用方法进行后台存库
  84.                         GuardianModelDao._save_update_users_info(user_id, user_name, real_name, email, tel, phone, org_id, org_array, comments)
  85.                         return wrap_response(SUCCESS_MSG, 200)    # 执行成功 返回操作成功信息及状态码 200
  86.                     except Exception as e:
  87.                         LOG.debug(e)
  88.                         abort(500)    # 否则 后台500 错误
  89.             else:
  90.                 abort(500)
  91.
后台函数

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

  1.  
  2.  @classmethod
  3.     def _save_update_users_info(cls, user_id, username, real_name, email, tel, phone, org_id, org_array, comments):
  4.         LOG.debug('start create users info...')
  5.         try:
  6.             with TxConn() as conn: # 连接数据库
  7.                 with conn.cursor() as cursor:
  8.                     # save user data
  9.                     now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
  10.                     # 接收变量赋值 user_data
  11.                     users_data = {
  12.                         "user_id":user_id,
  13.                         "username":username,
  14.                         "realname":real_name,
  15.                         "email":email,
  16.                         "tel":tel,
  17.                         "phone":phone,
  18.                         "org":org_id,
  19.                         "org_array":org_array,
  20.                         "comments":comments,
  21.                         "now_time":now_time
  22.                     }
  23.                     cursor.execute(cls.insert_users_info, users_data)    # 插入数据
  24.                 conn.commit()
  25.         except Exception as e:
  26.             LOG.exception('error saving users info', e)
  27.             raise e
  28.
测试脚本（新增用户）

python unittest做单元测试

  1.  
  2. import unittest, requests
  3.  
  4. class GuardianTestCase(unittest.TestCase):
  5.     guardian_base = 'http://127.0.0.1:7070'
  6.     user_token = keystone.user_token('default', 'tsbc', '123456') # 调用keystone生成登录用户的token
  7.     
  8. def test_create_user_api(self):
  9.         # 添加对应headers 及 tonken 用于数据传参和登录认证使用
  10.         myheaders = {'Content-Type': 'application/json'}
  11.         myheaders['X-Subject-Token'] = self.user_token
  12.         # 接口测试数据
  13.         user = {
  14.             "user": {
  15.                 "domain_id": "default",
  16.                 "enabled": True,
  17.                 "name": "tsbc1",
  18.                 "password": "123456",
  19.                 "realname": "Ray",
  20.                 "email": "tsbc@123.com",
  21.                 "tel":"+86-029-88880000",
  22.                 "phone": "13800138000",
  23.                 "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
  24.                 "org_array":None,
  25.                 "comments": "备注信息"
  26.             }
  27.         }
  28.         # 使用requests的post方法进行请求路由
  29.         _resp = requests.post(self.guardian_base + '/guardian/users', json=user, headers=myheaders, verify=True)
  30.         resp_json = _resp.content
  31.         # 对返回结果及状态进行校验
  32.         if _resp.ok:
  33.             self.assertIsNotNone(resp_json)
  34.             LOG.info('create users, response: {0}'.format(resp_json))
  35.         else:
  36.             LOG.error('error create users, response: {0}, status: {1}'.format(resp_json, _resp.status_code))
  37.
接口：修改用户

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

  1.  
  2. @app.route('/guardian/users/<string:user_id>', methods=['PATCH'])
  3. @require_role('guardian_admin')
  4. @login_required
  5. def update_user(user_id):
  6.     """
  7.     此处省略接收请求数据代码
  8.     """
  9.     try:
  10.             resp = usermgt.update_user(domain_id, user_id, username, user_passwd, project_id, enabled) # 更新keystone 用户密码、状态
  11.             if resp is not None:
  12.                 # 更新用户基本信息与新增是同一个接口，执行SQL时使用PostGresql特性，相同user_id数据只更新不会再插入
  13.                 GuardianModelDao._save_update_users_info(user_id, username, real_name, email, tel, phone, org_id, org_array, comments)
  14.                 return wrap_response(SUCCESS_MSG, 200)    # 更新成功返回 200
  15.         except Exception as e:
  16.             LOG.exception('error modify user', e)
  17.             abort(500)    # 更新失败500
  18.

测试脚本(修改用户)

  1.  
  2. def test_update_user_api(self):
  3.  
  4.         myheaders = {'Content-Type': 'application/json'}
  5.         myheaders['X-Subject-Token'] = self.user_token
  6.  
  7.         user = {
  8.             "user": {
  9.                 "domain_id": "default",
  10.                 "enabled": True,
  11.                 "name": "tsbc1",
  12.                 "password": "111111",
  13.                 "realname": "xioohua",
  14.                 "email": "haoydun@123.com",
  15.                 "tel": "+86-029-88880000",
  16.                 "phone": "13800138000",
  17.                 "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
  18.                 "comments": "备注信息"
  19.             }
  20.         }
  21.  
  22.         _resp = requests.patch(self.guardian_base + '/guardian/users/0d26176579e94d558c4c2d864b003241',   # 路由中添加要修改的用户ID requests使用 patch方法
  23.                                json=user, headers=myheaders, verify=True)
  24.         resp_json = _resp.content
  25.         # 返回值进行校验
  26.         if _resp.ok:
  27.             self.assertIsNotNone(resp_json)
  28.             LOG.info('update users, response: {0}'.format(resp_json))
  29.         else:
  30.             LOG.error('error update users, response: {0}, status: {1}'.format(resp_json, _resp.status_code))
  31.
接口：查询用户

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

  1.  
  2. @app.route('/guardian/users/<string:user_name>', methods=['GET'])    # 查询使用 GET 方法
  3. @require_role('guardian_admin')
  4. @login_required
  5. def query_user_by_username(user_name):
  6.  
  7.     try:
  8.         user_info = GuardianModelDao._query_users_info_by_name(user_name)    # 调用后台查询函数
  9.         if len(user_info):
  10.             return wrap_response(jsonpickle.dumps(user_info), 200)    # 成功返回200 并把返回内容转换成 json格式
  11.         else:
  12.             return wrap_response("[]", 200)   # 否则返回空
  13.     except Exception as e:
  14.         LOG.exception('error query user by name', e)
  15.         abort(500)
后台函数
  1.  
  2. @classmethod
  3.     def _query_users_info_by_id(cls, user_id):
  4.         LOG.debug('start list users info...')
  5.         users = []
  6.         with TxConn() as conn:
  7.             try:
  8.                 with conn.cursor() as cursor:
  9.                     data = (user_id,)
  10.                     cursor.execute(cls.query_users_info_by_id, data)
  11.                     # 查询结构如果是行数据，放进list中进行返回
  12.                     for (user_id, real_name, email, tel, phone, org_id, org_array, comments, create_time, update_time) in cursor:
  13.                         user = Users(user_id, real_name, email, tel, phone, org_id, org_array, comments, create_time, update_time)
  14.                         users.append(user)
  15.             except Exception as e:
  16.                 LOG.exception('error query users !')
  17.                 raise e
  18.         return users
  19.

测试脚本(查询用户)

  1.  
  2.  def test_query_user_api(self):
  3.  
  4.         myheaders = {'Content-Type': 'application/json'}
  5.         myheaders['X-Subject-Token'] = self.user_token
  6.  
  7.         _resp = requests.get(self.guardian_base + '/guardian/users/tsbc', headers=myheaders, verify=True) # 链接中加入用户名
  8.         resp_json = _resp.text
  9.         # 验证判断
  10.         if _resp.ok:
  11.             self.assertIsNotNone(resp_json)
  12.             LOG.info('query users, response: {0}'.format(resp_json))
  13.         else:
  14.             LOG.error('error query users, response: {0}, status: {1}'.format(resp_json, _resp.status_code))
接口：删除用户

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

  1.  
  2. @app.route('/guardian/users/<string:user_id>', methods=['DELETE']) # 传入用户id 使用 delete请求
  3. @require_role('guardian_admin')
  4. @login_required
  5. def delete_user(user_id):
  6.     if user_id is None:
  7.         abort(400)
  8.     try:
  9.         s_code = usermgt.delete_user(user_id)
  10.         if s_code == 204:
  11.             return wrap_response(SUCCESS_MSG, status=s_code) # 返回状态码
  12.     except Exception as e:
  13.         LOG.exception('error delete users info', e)
  14.         abort(500)

后台函数

  1.  
  2. def delete_user(user_id):
  3.     myheaders = {'Content-Type': 'application/json'}
  4.     myheaders['X-Auth-Token'] = settings.KEYSTONE_ADMIN_TOKEN
  5.     _resp = requests.delete(settings.KEYSTONE_BASE + '/v3/users/'+user_id,
  6.                             headers=myheaders, verify=False)
  7.  
  8.     if _resp.status_code == 204:
  9.         LOG.debug('delete users -- response {0}'.format(_resp))
  10.         return _resp.status_code
  11.     else:
  12.         LOG.error('error list users, response: {0}'.format(_resp))
  13.         raise KeystoneException('error delete user {0} with response status {1}'
  14.                                 .format(user_id, _resp.status_code), _resp.status_code)

测试脚本(删除用户)

  1.  
  2. def test_delete_users_api(self):
  3.  
  4.         myheaders = {'Content-Type': 'application/json'}
  5.         myheaders['X-Subject-Token'] = self.user_token
  6.         _resp = requests.delete(self.guardian_base + '/guardian/users/tsbc1', headers=myheaders, # 使用 delete 请求
  7.                                 verify=True)
  8.         resp_json = _resp.content
  9.         # 根据返回状态进行校验
  10.         if _resp.ok:
  11.             self.assertIsNotNone(resp_json)
  12.             LOG.info('delete users api, response: {0}'.format(resp_json))
  13.         else:
  14.             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:

  1.  
  2. {
  3.             "user": {
  4.                 "domain_id": "default",
  5.                 "enabled": True,
  6.                 "name": "tsbc1",
  7.                 "password": "111111",
  8.                 "realname": "Ray",
  9.                 "email": "haoydun@123.com",
  10.                 "tel":"+86-029-88880000",
  11.                 "phone": "13800138000",
  12.                 "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
  13.                 "org_array": null,
  14.                 "comments": "备注信息"
  15.             }
  16. }
##  UPDATE USER

1. URL:' /guardian/users/<string:user_id>'

2. METHOD: PATCH2. body:

  1.  
  2. {
  3.             "user": {
  4.                 "domain_id": "default",
  5.                 "enabled": True,
  6.                 "name": "tsbc1",
  7.                 "password": "111111",
  8.                 "realname": "Ray",
  9.                 "email": "tsbc@vip.qq.com",
  10.                 "tel":"+86-029-88880000",
  11.                 "phone": "13800138000",
  12.                 "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
  13.                 "org_array": null,
  14.                 "comments": "备注信息"
  15.             }
  16.   }

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

## 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

行动吧，在路上总比一直观望的要好，未来的你肯定会感 谢现在拼搏的自己！如果想学习提升找不到资料，没人答疑解惑时，请及时加入扣群： 320231853，里面有各种软件测试+开发资料和技术可以一起交流学习哦。

最后感谢每一个认真阅读我文章的人，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走：

​

这些资料，对于【软件测试】的朋友来说应该是最全面最完整的备战仓库，这个仓库也陪伴上万个测试工程师们走过最艰难的路程，希望也能帮助到你！

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/433815
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号