One connection per request?

[fengsp]

I have noticed that flask-rq does not use context-locals to save the redis connection, which means that everytime one request comes in, we are having a new connection.

What I need here is one connection manager so I do not have to handle connections myself.

Any reasons? So flask-rq is just a config manager? I do not see anything else meaningful here.

[howesb]

I managed to patch it with something like this

from flask_jwt import _jwt_required
from myflaskapp.auth import identity_handler

def mock_jwt_required(realm):
    return

def mock_identity_handler(payload):
    return User(123, '[email protected]', 'Foo', 'Bar')

@patch('flask_jwt._jwt_required', side_effect=mock_jwt_required)
@patch('myflaskapp.auth.identity_handler', side_effect=mock_identity_handler)
def test_my_authenticated_test_case(self, jwt_required_fn, identity_handler_fn):
    resp = self.app.get('/protected/resource')
    self.assertEquals(resp.status_code, 200, "status should be OK")

[jsamoocha]

@imtheoperator, I tried to do this, but am confronted with flask_jwt's current_identity being None in the method under test. I can see that mock_jwt_required is called, but mock_identity_handler in my test is never called. In your working example:

• Did your method under test actually use the current_identity local proxy?
• What version of flask_jwt did you use?

Thanks!

[howesb]

@jsamoocha I see what you mean. I wasn't using current_identity but I just created a new test and get the same behaviour as you:

Handler

@app.route('/api/v1/current-user', methods=['GET'])
@jwt_required()
def get_current_user():
    if not current_identity:
        return jsonify({"message": "Unauthorized"}), 401
    # etc ...

Test using mock patches in previous comment (for identity_handler)

# patch mocks as previous comment
def test_get_current_user(self, identity_handler_fn, jwt_required_fn):
    resp = self.app.get('/api/v1/current-user')
    self.assertEqual(resp.status_code, 200, "Status should be 200, was %s" % resp.status_code)

Gives

AssertionError: Status should be 200, was 401

So, yes, a solution would be useful!

[jsamoocha]

Ok, the following seems to be working now for a handler as mentioned above:

def mock_jwt_required(realm):
    return

# test
@patch('flask_jwt._jwt_required', side_effect=mock_jwt_required)
@patch('flask_jwt._request_ctx_stack')
    def test_get_user_profile(self, mock_request_ctx_stack, jwt_required_fn):
        mock_request_ctx_stack.top.current_identity = User(id=MOCK_USER_ID)
        # Any testing logic...

[jhonjairoroa87]

Just a small update given the the current flask version, note that:

• "jwt_required" has no underscode "_" at the begining
• Now is "current_user" instead of "current_identity"

def mock_jwt_required(realm):
    return

@patch('flask_jwt._jwt_required', side_effect=mock_jwt_required)
    @patch('flask_jwt._request_ctx_stack')
    def test_get_user_profile(self, mock_request_ctx_stack, jwt_required_fn):
        mock_request_ctx_stack.top.current_user = User(id=MOCK_USER_ID)
         # Any testing logic...

[LaundroMat]

For anyone who ran into a similar issue, it's better to patch the function the decorator is calling instead of trying to mock the decorator itself.

See my SO answer for details.

[jsamoocha]

What I currently find a useful alternative is to separate business logic from stuff that relies on the application context, e.g.

def _get(user_id):
    # whatever business logic to retrieve user data
    return user_service.find(user_id)

#endpoint
@app.route('/user)
@jwt_required()
def get():
    return _get(current_identity)

Now you only need to unit test _get(user_id) without bothering about any Flask or JWT state/side effects.