为了账号安全,请及时绑定邮箱和手机立即绑定

从另一个服务(微服务架构)验证 Flask 单元测试客户端?

从另一个服务(微服务架构)验证 Flask 单元测试客户端?

繁星点点滴滴 2022-01-05 10:43:16
问题:所以我的问题是我有一个 Flask 微服务想要对其实施单元测试,所以当我开始编写我的测试用例时,我发现我需要对单元测试客户端进行身份验证,因为某些端点需要授权,而整个身份验证的问题就出现了系统在另一个服务中该服务可以做的关于身份验证的所有工作是验证 JWT 令牌并从中获取用户 ID,所以这里是其中之一 views.pyfrom flask_restful import Resourcefrom common.decorators import authorizeclass PointsView(Resource):    decorators = [authorize]    def get(self, user):        result = {"points": user.active_points}        return result并授权装饰者来自 decorators.pyimport flaskimport jwtfrom jwt.exceptions import DecodeError, InvalidSignatureErrorfrom functools import wrapsfrom flask import requestfrom flask import current_app as appfrom app import dbfrom common.models import Userfrom common.utils import generate_error_responsedef authorize(f):    """This decorator for validate the logged in user """    @wraps(f)    def decorated_function(*args, **kwargs):        if 'Authorization' not in request.headers:            return "Unable to log in with provided credentials.", 403        raw_token = request.headers.get('Authorization')        if raw_token[0:3] != 'JWT':            return generate_error_response("Unable to log in with provided credentials.", 403)        token = str.replace(str(raw_token), 'JWT ', '')        try:            data = jwt_decode_handler(token)        except (DecodeError, InvalidSignatureError):            return generate_error_response("Unable to log in with provided credentials.", 403)        user = User.query.filter_by(id=int(data['user_id'])).first()        return f(user, *args, **kwargs)    return decorated_function和测试用例来自 tests.pyimport unittestfrom app import create_app, dbfrom common.models import Userclass TestMixin(object):    """    Methods to help all or most Test Cases    """    def __init__(self):        self.user = None我的身份验证系统的工作方式如下:任何服务(包括这个)请求用户服务来获取用户 JWT 令牌任何服务都将 JWT 令牌解码并从中获取用户 ID通过他的ID从数据库中获取用户对象所以我不知道如何在测试用例中进行身份验证。
查看完整描述

2 回答

?
茅侃侃

TA贡献1842条经验 获得超21个赞

这里只是一个例子。我跳过了一些小东西,例如create_app,jwt.decode(token)等等。我相信你能理解主要方法。结构:


src

├── __init__.py # empty

├── app.py

└── auth_example.py

应用程序.py:


from flask import Flask


from src.auth_example import current_identity, authorize


app = Flask(__name__)



@app.route('/')

@authorize()

def main():

    """

    You can use flask_restful - doesn't matter

    Do here all what you need:

        user = User.query.filter_by(id=int(current_identity['user_id'])).first()

        etc..

    just demo - return current user_id

    """


    return current_identity['user_id']

auth_example.py :


from flask import request, _request_ctx_stack

from functools import wraps

from werkzeug.local import LocalProxy


current_identity = LocalProxy(lambda: getattr(_request_ctx_stack.top, 'current_identity', None))



def jwt_decode_handler(token):

    """

    just do here all what you need. Should return current user data

    :param str token:

    :return: dict

    """

    # return jwt.decode(token), but now - just demo

    raise Exception('just demo')



def authorize():

    def _authorize(f):


        @wraps(f)

        def __authorize(*args, **kwargs):

            if 'Authorization' not in request.headers:

                return "Unable to log in with provided credentials.", 403


            raw_token = request.headers.get('Authorization')

            if raw_token[0:3] != 'JWT':

                return "Unable to log in with provided credentials.", 403

            token = str.replace(str(raw_token), 'JWT ', '')

            try:

                # I don't know do you use Flask-JWT or not

                # this is doesn't matter - all what you need is just to mock jwt_decode_handler result 

                _request_ctx_stack.top.current_identity = jwt_decode_handler(token)

            except Exception:

                return "Unable to log in with provided credentials.", 403


            return f(*args, **kwargs)


        return __authorize

    return _authorize

我们的测试:


import unittest


from mock import patch


from src.app import app


app.app_context().push()



class TestExample(unittest.TestCase):


    def test_main_403(self):

        # just a demo that @authorize works fine

        result = app.test_client().get('/')

        self.assertEqual(result.status_code, 403)


    def test_main_ok(self):

        expected = '1'

        # we say that jwt_decode_handler will return {'user_id': '1'}

        patcher = patch('src.auth_example.jwt_decode_handler', return_value={'user_id': expected})

        patcher.start()

        result = app.test_client().get(

            '/',

            # send a header to skip errors in the __authorize

            headers={

                'Authorization': 'JWT=blabla',

            },

        )

        # as you can see current_identity['user_id'] is '1' (so, it was mocked in view)

        self.assertEqual(result.data, expected)

        patcher.stop()

因此,在您的情况下,您只需要 mock jwt_decode_handler。另外我建议不要在装饰器中添加任何额外的参数。当您有两个以上具有不同参数、递归、硬处理等的装饰器时,将很难调试。


希望这可以帮助。


查看完整回答
反对 回复 2022-01-05
?
侃侃尔雅

TA贡献1801条经验 获得超16个赞

您能否在您的单元测试框架中创建一些模拟令牌(您的装饰器实际上可以像在真实请求中一样解码)并将它们与您的测试客户端一起发送?可以在此处查看其外观示例:https : //github.com/vimalloc/flask-jwt-extended/blob/master/tests/test_view_decorators.py#L321


查看完整回答
反对 回复 2022-01-05
  • 2 回答
  • 0 关注
  • 205 浏览
慕课专栏
更多

添加回答

举报

0/150
提交
取消
意见反馈 帮助中心 APP下载
官方微信