fix(gitlab): handle pagination of GitLab webhooks (reanahub#684)

Closes reanahub#682

reana_server/utils.py

@@ -17,7 +17,7 @@
 import secrets
 import sys
 import shutil
-from typing import Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union, Generator
 from uuid import UUID, uuid4
 
 import click
@@ -500,6 +500,19 @@ def _format_gitlab_secrets(gitlab_response):
     }
 
 
+def _unpaginate_gitlab_endpoint(url: str) -> Generator[Any, None, None]:
+    """Get all the paginated records of a given GitLab endpoint.
+
+    :param url: Endpoint URL to the first page.
+    """
+    while url:
+        logging.debug(f"Request to '{url}' while unpaginating GitLab endpoint")
+        response = requests.get(url)
+        response.raise_for_status()
+        yield from response.json()
+        url = response.links.get("next", {}).get("url")
+
+
 def _get_gitlab_hook_id(project_id, gitlab_token):
     """Return REANA hook id from a GitLab project if it is connected.
 
@@ -511,27 +524,22 @@ def _get_gitlab_hook_id(project_id, gitlab_token):
     """
     gitlab_hooks_url = (
         REANA_GITLAB_URL
-        + "/api/v4/projects/{0}/hooks?access_token={1}".format(project_id, gitlab_token)
+        + "/api/v4/projects/{0}/hooks?per_page=100&access_token={1}".format(
+            project_id, gitlab_token
+        )
     )
-    response = requests.get(gitlab_hooks_url)
+    create_workflow_url = url_for("workflows.create_workflow", _external=True)
 
-    if not response.ok:
+    try:
+        for hook in _unpaginate_gitlab_endpoint(gitlab_hooks_url):
+            if hook["url"] and hook["url"] == create_workflow_url:
+                return hook["id"]
+    except requests.HTTPError as e:
         logging.warning(
-            f"GitLab hook request failed with status code: {response.status_code}, "
-            f"content: {response.content}"
+            f"GitLab hook request failed with status code: {e.response.status_code}, "
+            f"content: {e.response.content}"
         )
-        return None
-
-    response_json = response.json()
-    create_workflow_url = url_for("workflows.create_workflow", _external=True)
-    return next(
-        (
-            hook["id"]
-            for hook in response_json
-            if hook["url"] and hook["url"] == create_workflow_url
-        ),
-        None,
-    )
+    return None
 
 
 class RequestStreamWithLen(object):

tests/test_utils.py

@@ -1,17 +1,23 @@
 # This file is part of REANA.
-# Copyright (C) 2021, 2022, 2023 CERN.
+# Copyright (C) 2021, 2022, 2023, 2024 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """REANA-Server tests for utils module."""
 
 import pathlib
+from unittest.mock import call, patch, Mock
 import pytest
 
 from reana_commons.errors import REANAValidationError
 from reana_db.models import UserToken, UserTokenStatus, UserTokenType
-from reana_server.utils import is_valid_email, filter_input_files, get_user_from_token
+from reana_server.utils import (
+    is_valid_email,
+    filter_input_files,
+    get_user_from_token,
+    _unpaginate_gitlab_endpoint,
+)
 
 
 @pytest.mark.parametrize(
@@ -81,3 +87,25 @@ def test_get_user_from_token_two_tokens(default_user, session):
     # Check that old revoked token does not work
     with pytest.raises(ValueError, match="revoked"):
         get_user_from_token(old_token.token)
+
+
+@patch("requests.get")
+def test_gitlab_pagination(mock_get):
+    """Test getting all paginated results from GitLab."""
+    # simulating two pages
+    first_response = Mock()
+    first_response.ok = True
+    first_response.links = {"next": {"url": "next_url"}}
+    first_response.json.return_value = [1, 2]
+
+    second_response = Mock()
+    second_response.ok = True
+    second_response.links = {}
+    second_response.json.return_value = [3, 4]
+
+    mock_get.side_effect = [first_response, second_response]
+
+    res = list(_unpaginate_gitlab_endpoint("first_url"))
+
+    assert res == [1, 2, 3, 4]
+    assert mock_get.call_args_list == [call("first_url"), call("next_url")]

tests/test_views.py

@@ -835,6 +835,7 @@ def test_gitlab_projects(app: Flask, default_user):
         mock_response_webhook = Mock()
         mock_response_webhook.ok = True
         mock_response_webhook.status_code = 200
+        mock_response_webhook.links = {}
         mock_response_webhook.json.return_value = [
             {"id": 1234, "url": "wrong_url"},
             {