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": "[e-mail]",
                "tel": "[phone]",
                "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', '123456') # 調用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": "123456",
                "realname": "Ray",
                "email": "[e-mail]",
                "tel":"[phone]",
                "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: {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)    # 更新成功返回 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": "111111",
                "realname": "xioohua",
                "email": "[e-mail]",
                "tel": "[phone]",
                "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: {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)

后台函數

@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/<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, 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": "111111",
                "realname": "Ray",
                "email": "[e-mail]",
                "tel":"[phone]",
                "phone": "13800138000",
                "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
                "org_array": null,
                "comments": "備注信息"
            }
}

##  UPDATE USER

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

2. METHOD: PATCH

2. body:

{
            "user": {
                "domain_id": "default",
                "enabled": True,
                "name": "tsbc1",
                "password": "111111",
                "realname": "Ray",
                "email": "[e-mail]",
                "tel":"[phone]",
                "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