Support triggering commands via review assignment in Gitlab

pr_agent/servers/gitlab_webhook.py

@@ -1,4 +1,6 @@
+import asyncio
 import copy
+import hashlib
 import json
 import os
 import re
@@ -108,6 +110,87 @@ def is_draft_ready(data) -> bool:
         get_logger().error(f"Failed 'is_draft_ready' logic: {e}")
     return False
 
+_bot_user_id_cache = {}
+
+async def _get_bot_user_id():
+    gitlab_url = get_settings().get("GITLAB.URL", "https://gitlab.com")
+    gitlab_token = get_settings().get("GITLAB.PERSONAL_ACCESS_TOKEN", None)
+    if not gitlab_token:
+        get_logger().error("No GitLab token available for bot user ID resolution")
+        return None
+
+    cache_key = hashlib.sha256(f"{gitlab_url}:{gitlab_token}".encode()).hexdigest()
+
+    cached = _bot_user_id_cache.get(cache_key)
+    if cached is not None:
+        return cached
+
+    def _resolve_sync():
+        import gitlab
+
+        ssl_verify = get_settings().get("GITLAB.SSL_VERIFY", True)
+        if isinstance(ssl_verify, str) and ssl_verify.lower() in ("true", "false"):
+            ssl_verify = ssl_verify.lower() == "true"
+
+        auth_method = get_settings().get("GITLAB.AUTH_TYPE", "oauth_token")
+        if auth_method not in ("oauth_token", "private_token"):
+            raise ValueError(
+                f"Unsupported GITLAB.AUTH_TYPE: '{auth_method}'. "
+                f"Must be 'oauth_token' or 'private_token'."
+            )
+
+        if auth_method == "oauth_token":
+            gl = gitlab.Gitlab(
+                url=gitlab_url,
+                oauth_token=gitlab_token,
+                ssl_verify=ssl_verify
+            )
+        else:
+            gl = gitlab.Gitlab(
+                url=gitlab_url,
+                private_token=gitlab_token,
+                ssl_verify=ssl_verify
+            )
+        gl.auth()
+        return gl.user.id
+
+    try:
+        user_id = await asyncio.to_thread(_resolve_sync)
+        if len(_bot_user_id_cache) > 1000:
+            _bot_user_id_cache.clear()
+        _bot_user_id_cache[cache_key] = user_id
+        get_logger().info(f"Bot user ID resolved via API: {user_id}")
+        return user_id
+    except Exception as e:
+        get_logger().error(f"Failed to resolve bot user ID: {e}")
+        return None
+
+async def is_bot_assigned_as_reviewer(data) -> bool:
+    try:
+        changes = data.get("changes")
+        if not isinstance(changes, dict):
+            return False
+        if "reviewers" not in changes:
+            return False
+        reviewers_change = changes["reviewers"]
+        if not isinstance(reviewers_change, dict):
+            return False
+        previous = reviewers_change.get("previous")
+        if not isinstance(previous, list):
+            previous = []
+        current = reviewers_change.get("current")
+        if not isinstance(current, list):
+            current = []
+        bot_user_id = await _get_bot_user_id()
+        if bot_user_id is None:
+            return False
+        previous_ids = {r.get("id") for r in previous if isinstance(r, dict)}
+        current_ids = {r.get("id") for r in current if isinstance(r, dict)}
+        return bot_user_id in current_ids and bot_user_id not in previous_ids
+    except Exception as e:
+        get_logger().error(f"Failed 'is_bot_assigned_as_reviewer' logic: {e}")
+    return False
+
 def should_process_pr_logic(data) -> bool:
     try:
         if not data.get('object_attributes', {}):
@@ -217,7 +300,9 @@ async def inner(data: dict):
             # ignore MRs based on title, labels, source and target branches
             if not should_process_pr_logic(data):
                 return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
-            object_attributes = data.get('object_attributes', {})
+            object_attributes = data.get('object_attributes')
+            if not isinstance(object_attributes, dict):
+                object_attributes = {}
             if object_attributes.get('action') in ['open', 'reopen']:
                 url = object_attributes.get('url')
                 get_logger().info(f"New merge request: {url}")
@@ -247,7 +332,7 @@ async def inner(data: dict):
 
                 get_logger().debug(f'A push event has been received: {url}')
                 await _perform_commands_gitlab("push_commands", PRAgent(), url, log_context, data)
-                
+
             # for draft to ready triggered merge requests
             elif object_attributes.get('action') == 'update' and is_draft_ready(data):
                 url = object_attributes.get('url')
@@ -256,6 +341,44 @@ async def inner(data: dict):
                 # same as open MR
                 await _perform_commands_gitlab("pr_commands", PRAgent(), url, log_context, data)
 
+            # for reviewer assignment triggered merge requests
+            elif object_attributes.get('action') == 'update' and not object_attributes.get('oldrev'):
+                url = object_attributes.get('url')
+                if not url:
+                    return JSONResponse(status_code=status.HTTP_200_OK,
+                                        content=jsonable_encoder({"message": "success"}))
+
+                # Fast early-exit: no reviewer changes means nothing to do
+                changes = data.get("changes")
+                if not isinstance(changes, dict) or "reviewers" not in changes:
+                    return JSONResponse(status_code=status.HTTP_200_OK,
+                                        content=jsonable_encoder({"message": "success"}))
+
+                apply_repo_settings(url)
+                handle_assignment = get_settings().gitlab.get("handle_reviewer_assignment", False)
+                if isinstance(handle_assignment, str):
+                    handle_assignment = handle_assignment.lower() in ("true", "1", "yes")
+                if not handle_assignment:
+                    return JSONResponse(status_code=status.HTTP_200_OK,
+                                        content=jsonable_encoder({"message": "success"}))
+
+                # Check PR logic after applying repo settings
+                if not should_process_pr_logic(data):
+                    return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
+
+                if is_draft(data):
+                    get_logger().info(f"Skipping draft MR reviewer assignment: {url}")
+                    return JSONResponse(status_code=status.HTTP_200_OK,
+                                        content=jsonable_encoder({"message": "success"}))
+                if await is_bot_assigned_as_reviewer(data):
+                    reviewer_commands = get_settings().gitlab.get("reviewer_commands", [])
+                    if not isinstance(reviewer_commands, list) or not all(isinstance(c, str) for c in reviewer_commands):
+                        get_logger().warning("gitlab.reviewer_commands is not a list of strings, skipping")
+                        return JSONResponse(status_code=status.HTTP_200_OK,
+                                            content=jsonable_encoder({"message": "success"}))
+                    get_logger().info(f"Bot was assigned as reviewer on MR: {url}")
+                    await _perform_commands_gitlab("reviewer_commands", PRAgent(), url, log_context, data)
+
         elif data.get('object_kind') == 'note' and data.get('event_type') == 'note': # comment on MR
             if 'merge_request' in data:
                 mr = data['merge_request']

pr_agent/settings/configuration.toml

@@ -269,6 +269,11 @@ push_commands = [
 ]
 # Configure SSL validation for GitLab. Can be either set to the path of a custom CA or disabled entirely.
 # ssl_verify = true
+# Auto-trigger commands when the bot is assigned as a reviewer on an MR
+handle_reviewer_assignment = false
+reviewer_commands = [
+    "/review",
+]
 
 [gitea]
 url = "https://gitea.com"

tests/unittest/test_gitlab_reviewer_assignment.py

@@ -0,0 +1,180 @@
+import asyncio
+from unittest import mock
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def setup_env(monkeypatch):
+    monkeypatch.setenv("GITLAB__URL", "https://gitlab.example.com")
+
+
+class TestIsBotAssignedAsReviewer:
+    BOT_ID = 516
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_detects_new_assignment(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = self.BOT_ID
+        data = {
+            "changes": {
+                "reviewers": {
+                    "previous": [],
+                    "current": [{"id": self.BOT_ID, "username": "k2so-bot"}],
+                }
+            }
+        }
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is True
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_ignores_already_assigned(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = self.BOT_ID
+        data = {
+            "changes": {
+                "reviewers": {
+                    "previous": [{"id": self.BOT_ID, "username": "k2so-bot"}],
+                    "current": [{"id": self.BOT_ID, "username": "k2so-bot"}],
+                }
+            }
+        }
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_no_reviewers_key(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = self.BOT_ID
+        data = {"changes": {"updated_at": {"previous": "old", "current": "new"}}}
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    def test_changes_not_dict(self):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        data = {"changes": "not-a-dict"}
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_reviewers_not_dict(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = self.BOT_ID
+        data = {"changes": {"reviewers": "not-a-dict"}}
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    def test_no_changes_key(self):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        data = {"object_kind": "merge_request"}
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_bot_id_unresolvable(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = None
+        data = {
+            "changes": {
+                "reviewers": {
+                    "previous": [],
+                    "current": [{"id": self.BOT_ID}],
+                }
+            }
+        }
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_previous_with_non_dict_entries(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = self.BOT_ID
+        data = {
+            "changes": {
+                "reviewers": {
+                    "previous": [{"id": 100}, "not-a-dict"],
+                    "current": [{"id": self.BOT_ID}],
+                }
+            }
+        }
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is True
+
+
+class TestGetBotUserId:
+    @staticmethod
+    def _make_fake_gitlab(user_id):
+        fake = mock.MagicMock()
+        fake.Gitlab.return_value.auth.return_value = None
+        fake.Gitlab.return_value.user.id = user_id
+        return fake
+
+    @staticmethod
+    def _make_settings(url, token):
+        s = mock.MagicMock()
+        s.get.side_effect = lambda k, d=None: {
+            "GITLAB.URL": url,
+            "GITLAB.PERSONAL_ACCESS_TOKEN": token,
+            "GITLAB.SSL_VERIFY": True,
+            "GITLAB.AUTH_TYPE": "oauth_token",
+        }.get(k, d)
+        return s
+
+    def test_caches_by_credential(self):
+        from pr_agent.servers.gitlab_webhook import _bot_user_id_cache, _get_bot_user_id
+        _bot_user_id_cache.clear()
+
+        with mock.patch("pr_agent.servers.gitlab_webhook.get_settings",
+                        return_value=self._make_settings("https://a.example.com", "token-a")):
+            with mock.patch.dict("sys.modules", {"gitlab": self._make_fake_gitlab(111)}):
+                assert asyncio.run(_get_bot_user_id()) == 111
+
+        with mock.patch("pr_agent.servers.gitlab_webhook.get_settings",
+                        return_value=self._make_settings("https://a.example.com", "token-b")):
+            with mock.patch.dict("sys.modules", {"gitlab": self._make_fake_gitlab(222)}):
+                assert asyncio.run(_get_bot_user_id()) == 222
+
+        assert len(_bot_user_id_cache) >= 2
+
+    def test_no_cache_on_failure(self):
+        from pr_agent.servers.gitlab_webhook import _bot_user_id_cache, _get_bot_user_id
+        _bot_user_id_cache.clear()
+
+        fake = self._make_fake_gitlab(0)
+        fake.Gitlab.side_effect = RuntimeError("auth failed")
+
+        with mock.patch("pr_agent.servers.gitlab_webhook.get_settings",
+                        return_value=self._make_settings("https://x.example.com", "fail-token")):
+            with mock.patch.dict("sys.modules", {"gitlab": fake}):
+                assert asyncio.run(_get_bot_user_id()) is None
+
+        assert len(_bot_user_id_cache) == 0
+
+    def test_respects_auth_type_private_token(self):
+        from pr_agent.servers.gitlab_webhook import _bot_user_id_cache, _get_bot_user_id
+        _bot_user_id_cache.clear()
+
+        s = mock.MagicMock()
+        s.get.side_effect = lambda k, d=None: {
+            "GITLAB.URL": "https://x.example.com",
+            "GITLAB.PERSONAL_ACCESS_TOKEN": "tok",
+            "GITLAB.SSL_VERIFY": True,
+            "GITLAB.AUTH_TYPE": "private_token",
+        }.get(k, d)
+
+        fake_gitlab = self._make_fake_gitlab(99)
+
+        with mock.patch("pr_agent.servers.gitlab_webhook.get_settings", return_value=s):
+            with mock.patch.dict("sys.modules", {"gitlab": fake_gitlab}):
+                assert asyncio.run(_get_bot_user_id()) == 99
+
+        call_kwargs = fake_gitlab.Gitlab.call_args.kwargs
+        assert "private_token" in call_kwargs
+        assert call_kwargs["private_token"] == "tok"
+
+    def test_no_token_returns_none(self):
+        from pr_agent.servers.gitlab_webhook import _bot_user_id_cache, _get_bot_user_id
+        _bot_user_id_cache.clear()
+
+        s = mock.MagicMock()
+        s.get.side_effect = lambda k, d=None: {
+            "GITLAB.URL": "https://x.example.com",
+            "GITLAB.PERSONAL_ACCESS_TOKEN": None,
+            "GITLAB.SSL_VERIFY": True,
+            "GITLAB.AUTH_TYPE": "oauth_token",
+        }.get(k, d)
+
+        with mock.patch("pr_agent.servers.gitlab_webhook.get_settings", return_value=s):
+            assert asyncio.run(_get_bot_user_id()) is None