Adding queue to bugzilla webhook route (#939)

* Adding queue logic to bugzilla webhook. Still need tests.
* Adding test scenarios for execute_or_queue
---------
Co-authored-by: Mathieu Leplatre <[email protected]>
Co-authored-by: Graham Beckley <[email protected]>

Author: 3 people

.gitignore

@@ -127,3 +127,7 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# IDEs
+.idea/
+.vscode/

jbi/router.py

@@ -16,7 +16,8 @@
 from jbi.configuration import ACTIONS
 from jbi.environment import Settings, get_settings
 from jbi.models import Actions
-from jbi.runner import IgnoreInvalidRequestError, execute_action
+from jbi.queue import DeadLetterQueue, get_dl_queue
+from jbi.runner import execute_or_queue
 
 SettingsDep = Annotated[Settings, Depends(get_settings)]
 ActionsDep = Annotated[Actions, Depends(lambda: ACTIONS)]
@@ -64,17 +65,14 @@ def api_key_auth(
     "/bugzilla_webhook",
     dependencies=[Depends(api_key_auth)],
 )
-def bugzilla_webhook(
+async def bugzilla_webhook(
     request: Request,
     actions: ActionsDep,
+    queue: Annotated[DeadLetterQueue, Depends(get_dl_queue)],
     webhook_request: bugzilla.WebhookRequest = Body(..., embed=False),
 ):
     """API endpoint that Bugzilla Webhook Events request"""
-    try:
-        result = execute_action(webhook_request, actions)
-        return result
-    except IgnoreInvalidRequestError as exception:
-        return {"error": str(exception)}
+    return await execute_or_queue(webhook_request, queue, actions)
 
 
 @router.get(

jbi/runner.py

@@ -23,6 +23,7 @@
     JiraContext,
     RunnerContext,
 )
+from jbi.queue import DeadLetterQueue
 from jbi.steps import StepStatus
 
 logger = logging.getLogger(__name__)
@@ -157,6 +158,29 @@ def __call__(self, context: ActionContext) -> ActionResult:
         return True, {"responses": responses}
 
 
+async def execute_or_queue(
+    request: bugzilla.WebhookRequest, queue: DeadLetterQueue, actions: Actions
+):
+    if await queue.is_blocked(request):
+        # If it's blocked, store it and wait for it to be processed later.
+        await queue.postpone(request)
+        logger.info(
+            "%r event on Bug %s was put in queue for later processing.",
+            request.event.action,
+            request.bug.id,
+            extra={"payload": request.model_dump()},
+        )
+        return {"status": "skipped"}
+
+    try:
+        return execute_action(request, actions)
+    except IgnoreInvalidRequestError as exc:
+        return {"status": "invalid", "error": str(exc)}
+    except Exception as exc:
+        await queue.track_failed(request, exc)
+        return {"status": "failed", "error": str(exc)}
+
+
 @statsd.timer("jbi.action.execution.timer")
 def execute_action(
     request: bugzilla.WebhookRequest,

tests/conftest.py

@@ -15,6 +15,7 @@
 from jbi import Operation, bugzilla, jira
 from jbi.environment import Settings
 from jbi.models import ActionContext
+from jbi.queue import DeadLetterQueue
 
 
 class FilteredLogCaptureFixture(pytest.LogCaptureFixture):
@@ -94,7 +95,9 @@ def test_api_key():
 @pytest.fixture
 def authenticated_client(app, test_api_key):
     """An test client with a valid API key."""
-    return TestClient(app, headers={"X-Api-Key": test_api_key})
+    return TestClient(
+        app, headers={"X-Api-Key": test_api_key}, raise_server_exceptions=False
+    )
 
 
 @pytest.fixture
@@ -108,6 +111,11 @@ def actions(actions_factory):
     return actions_factory()
 
 
+@pytest.fixture
+def mock_queue():
+    return mock.MagicMock(spec=DeadLetterQueue)
+
+
 @pytest.fixture(autouse=True)
 def mocked_bugzilla(request):
     if "no_mocked_bugzilla" in request.keywords:

tests/unit/test_app.py

@@ -79,9 +79,10 @@ def test_traces_sampler(sampling_context, expected):
     assert traces_sampler(sampling_context) == expected
 
 
-def test_errors_are_reported_to_sentry(anon_client, bugzilla_webhook_request):
+@pytest.mark.asyncio
+async def test_errors_are_reported_to_sentry(anon_client, bugzilla_webhook_request):
     with patch("sentry_sdk.hub.Hub.capture_event") as mocked:
-        with patch("jbi.router.execute_action", side_effect=ValueError):
+        with patch("jbi.router.execute_or_queue", side_effect=ValueError):
             with pytest.raises(ValueError):
                 anon_client.post(
                     "/bugzilla_webhook",
@@ -92,7 +93,8 @@ def test_errors_are_reported_to_sentry(anon_client, bugzilla_webhook_request):
     assert mocked.called, "Sentry captured the exception"
 
 
-def test_request_id_is_passed_down_to_logger_contexts(
+@pytest.mark.asyncio
+async def test_request_id_is_passed_down_to_logger_contexts(
     caplog,
     bugzilla_webhook_request,
     authenticated_client,

tests/unit/test_retry.py

@@ -3,7 +3,6 @@
 
 import pytest
 
-from jbi.queue import DeadLetterQueue
 from jbi.retry import RETRY_TIMEOUT_DAYS, retry_failed
 from jbi.runner import execute_action
 
@@ -20,11 +19,6 @@ async def aiter_sync(iterable):
         yield i
 
 
-@pytest.fixture
-def mock_queue():
-    return MagicMock(spec=DeadLetterQueue)
-
-
 @pytest.fixture
 def mock_executor():
     return MagicMock(spec=execute_action)

tests/unit/test_router.py

@@ -8,6 +8,7 @@
 from fastapi.testclient import TestClient
 
 from jbi.environment import get_settings
+from jbi.queue import get_dl_queue
 
 
 def test_read_root(anon_client):
@@ -156,6 +157,35 @@ def test_webhook_is_200_if_action_raises_IgnoreInvalidRequestError(
     assert response.json()["error"] == "no bug whiteboard matching action tags: devtest"
 
 
+def test_webhook_is_200_if_action_raises_Exception(
+    webhook_request_factory, mocked_bugzilla, authenticated_client
+):
+    webhook = webhook_request_factory()
+    mocked_bugzilla.get_bug.side_effect = Exception("Throwing an exception")
+
+    response = authenticated_client.post(
+        "/bugzilla_webhook",
+        data=webhook.model_dump_json(),
+    )
+    assert response
+    assert response.status_code == 200
+
+
+def test_webhook_is_500_if_queue_raises_Exception(
+    webhook_request_factory, mocked_bugzilla, authenticated_client, mock_queue
+):
+    webhook = webhook_request_factory()
+    authenticated_client.app.dependency_overrides[get_dl_queue] = lambda: mock_queue
+    mock_queue.is_blocked.side_effect = Exception("Throwing an exception")
+
+    response = authenticated_client.post(
+        "/bugzilla_webhook",
+        data=webhook.model_dump_json(),
+    )
+    assert response
+    assert response.status_code == 500
+
+
 def test_webhook_is_401_if_unathenticated(
     webhook_request_factory, mocked_bugzilla, anon_client
 ):

tests/unit/test_runner.py

@@ -9,7 +9,13 @@
 from jbi.environment import get_settings
 from jbi.errors import ActionNotFoundError, IgnoreInvalidRequestError
 from jbi.models import ActionContext
-from jbi.runner import Executor, execute_action, lookup_action
+from jbi.runner import (
+    Actions,
+    Executor,
+    execute_action,
+    execute_or_queue,
+    lookup_action,
+)
 
 
 def test_bugzilla_object_is_always_fetched(
@@ -249,6 +255,55 @@ def test_runner_ignores_request_if_jira_is_linked_but_without_whiteboard(
     assert str(exc_info.value) == "no bug whiteboard matching action tags: devtest"
 
 
+@pytest.mark.asyncio
+async def test_execute_or_queue_happy_path(
+    mock_queue,
+    bugzilla_webhook_request,
+):
+    mock_queue.is_blocked.return_value = False
+    await execute_or_queue(
+        request=bugzilla_webhook_request,
+        queue=mock_queue,
+        actions=mock.MagicMock(spec=Actions),
+    )
+    mock_queue.is_blocked.assert_called_once()
+    mock_queue.postpone.assert_not_called()
+    mock_queue.track_failed.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_execute_or_queue_blocked(
+    actions,
+    mock_queue,
+    bugzilla_webhook_request,
+):
+    mock_queue.is_blocked.return_value = True
+    await execute_or_queue(
+        request=bugzilla_webhook_request,
+        queue=mock_queue,
+        actions=mock.MagicMock(spec=Actions),
+    )
+    mock_queue.is_blocked.assert_called_once()
+    mock_queue.postpone.assert_called_once()
+    mock_queue.track_failed.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_execute_or_queue_exception(
+    actions,
+    mock_queue,
+    bugzilla_webhook_request,
+):
+    mock_queue.is_blocked.return_value = False
+    # should trigger an exception for this scenario
+    await execute_or_queue(
+        request=bugzilla_webhook_request, queue=mock_queue, actions=actions
+    )
+    mock_queue.is_blocked.assert_called_once()
+    mock_queue.postpone.assert_not_called()
+    mock_queue.track_failed.assert_called_once()
+
+
 def test_default_invalid_init():
     with pytest.raises(TypeError):
         Executor()