从另一个服务(微服务体系结构)对 Flask 单元测试客户端进行身份验证



问题:

所以我的问题是我有一个 Flask 微服务想要对其实施单元测试,所以当我开始编写测试用例时,我发现我需要对单元测试客户端进行身份验证,因为某些端点需要授权,问题来了,整个身份验证系统在另一个服务中,该服务可以对身份验证所做的一切都是验证 JWT 令牌并从中获取用户 ID这是views.py之一

from flask_restful import Resource
from common.decorators import authorize

class PointsView(Resource):
    decorators = [authorize]
    def get(self, user):
        result = {"points": user.active_points}
        return result

并授权装饰师从decorators.py

import flask
import jwt
from jwt.exceptions import DecodeError, InvalidSignatureError
from functools import wraps
from flask import request
from flask import current_app as app
from app import db
from common.models import User
from common.utils import generate_error_response
def 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.py的测试用例

import unittest
from app import create_app, db
from common.models import User

class TestMixin(object):
    """
    Methods to help all or most Test Cases
    """
    def __init__(self):
        self.user = None
    """ User Fixture for testing """
    def user_test_setup(self):
        self.user = User(
            username="user1",
            active_points=0
        )
        db.session.add(self.user)
        db.session.commit()
    def user_test_teardown(self):
        db.session.query(User).delete()
        db.session.commit()

class PointsTestCase(unittest.TestCase, TestMixin):
    """This class represents the points test case"""
    def setUp(self):
        """Define test variables and initialize app."""
        self.app = create_app("testing")
        self.client = self.app.test_client
        with self.app.app_context():
            self.user_test_setup()
    def test_get_points(self):
        """Test API can create a points (GET request)"""
        res = self.client().get('/user/points/')
        self.assertEqual(res.status_code, 200)
        self.assertEquals(res.data, {"active_points": 0})
    def tearDown(self):
        with self.app.app_context():
            self.user_test_teardown()

# Make the tests conveniently executable
if __name__ == "__main__":
    unittest.main()

我的身份验证系统的工作方式如下:

  1. 任何服务(包括此服务(请求用户服务以获取用户 JWT令 牌
  2. 任何服务都采用解码的 JWT 令牌并获取用户 ID从它
  3. 通过用户的 ID 从数据库中获取用户对象

所以我不知道如何在测试用例中使身份验证流程。

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

src
├── __init__.py # empty
├── app.py
└── auth_example.py

app.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()

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

希望这有帮助。

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

相关内容

最新更新