style: apply consistent formatting across the jwt changes

Author: tomondre

reana_server/decorators.py

@@ -38,8 +38,8 @@ def wrapper(*args, **kwargs):
                     user = get_user_from_token(request.headers["X-Gitlab-Token"])
                 elif "access_token" in request.args:
                     user = get_user_from_token(request.args.get("access_token"))
-                elif include_jwt and request.headers['Authorization']:
-                    user = _get_user_from_jwt(request.headers['Authorization'])
+                elif include_jwt and request.headers["Authorization"]:
+                    user = _get_user_from_jwt(request.headers["Authorization"])
 
                 if not user:
                     return jsonify(message="User not signed in"), 401

reana_server/utils.py

@@ -434,12 +434,14 @@ def _get_user_from_invenio_user(id):
         raise ValueError("User access token revoked.")
     return user
 
+
 def _get_user_by_idpid(idp_id):
     user = Session.query(User).filter_by(idp_id=idp_id).one_or_none()
     if not user:
         raise ValueError("No users registered with this id")
     return user
 
+
 def _get_reana_yaml_from_gitlab(webhook_data, user_id):
     reana_yaml = "reana.yaml"
     if webhook_data["object_kind"] == "push":
@@ -676,7 +678,7 @@ def fetch_and_parse_jwk():
     Raises:
         ValueError: If JWK fetch fails or no matching key found
     """
-    if not hasattr(fetch_and_parse_jwk, '_cache'):
+    if not hasattr(fetch_and_parse_jwk, "_cache"):
         fetch_and_parse_jwk._cache = None
 
     if not fetch_and_parse_jwk._cache:
@@ -690,7 +692,6 @@ def fetch_and_parse_jwk():
     return jwks
 
 
-
 def _get_user_from_jwt(header: str) -> User:
     """Get user from JWT token.
 
@@ -704,7 +705,7 @@ def _get_user_from_jwt(header: str) -> User:
         ValueError: If token is invalid or user not found
     """
     try:
-        if not header.startswith('Bearer '):
+        if not header.startswith("Bearer "):
             raise ValueError("Invalid authorization header format")
 
         token = header.split(" ")[1]
@@ -715,7 +716,7 @@ def _get_user_from_jwt(header: str) -> User:
         claims = jwt.decode(token, key_set)
         claims.validate()
 
-        idp_id = claims.get('sub')
+        idp_id = claims.get("sub")
         if not idp_id:
             raise ValueError("Token missing subject claim")
 

tests/test_decorators.py

@@ -32,6 +32,7 @@ def test_signing_required_with_token(user0: User):
         assert code == 401
         assert error["message"] == "User has no active tokens"
 
+
 def test_signin_required_with_jwt(user0: User):
     """Test `signin_required` with JWT token authentication."""
     mock_endpoint = Mock(return_value=(jsonify(message="Success"), 200))
@@ -54,7 +55,10 @@ def test_signin_required_with_jwt(user0: User):
                 # Should call the endpoint since authentication succeeded
                 mock_endpoint.assert_called_once()
                 assert code == 200
-                assert json.loads(response.get_data(as_text=True))["message"] == "Success"
+                assert (
+                    json.loads(response.get_data(as_text=True))["message"] == "Success"
+                )
+
 
 def test_signin_required_with_invalid_jwt(user0: User):
     """Test `signin_required` with invalid JWT token."""
@@ -70,12 +74,18 @@ def test_signin_required_with_invalid_jwt(user0: User):
     mock_request_context.__enter__.return_value = mock_request
 
     with patch("reana_server.decorators.current_user", mock_current_user):
-        with patch("reana_server.decorators._get_user_from_jwt", side_effect=ValueError("Invalid token")):
+        with patch(
+            "reana_server.decorators._get_user_from_jwt",
+            side_effect=ValueError("Invalid token"),
+        ):
             with patch("reana_server.decorators.request", mock_request):
                 decorated_endpoint = signin_required(include_jwt=True)(mock_endpoint)
                 response, code = decorated_endpoint()
 
                 # Should not call the endpoint since authentication failed
                 mock_endpoint.assert_not_called()
                 assert code == 403
-                assert json.loads(response.get_data(as_text=True))["message"] == "Invalid token"
+                assert (
+                    json.loads(response.get_data(as_text=True))["message"]
+                    == "Invalid token"
+                )