T001: notify GitLab MR approval changes

gitlab_bot.py

@@ -17,6 +17,7 @@
 
 from dotenv import load_dotenv
 
+from src.approval_notification import ApprovalNotificationHooks, LogChannel
 from src.config import (
     bot_gitlab_token,
     bot_gitlab_url,
@@ -57,6 +58,7 @@ def _load_gitlab_bot():
 issue_hooks = IssueHooks()
 merge_request_hooks = MergeRequestHooks()
 note_hooks = NoteHooks()
+approval_notification_hooks = ApprovalNotificationHooks(LogChannel())
 
 
 @bot.router.register("Issue Hook", action="open")
@@ -101,6 +103,18 @@ async def merge_request_reopen_event(event, gl, *args, **kwargs):
         await merge_request_hooks.merge_request_reopen_event(event, gl, args, kwargs)
 
 
+@bot.router.register("Merge Request Hook", action="approved")
+@bot.router.register("Merge Request Hook", action="approval")
+async def merge_request_approval_event(event, gl, *args, **kwargs):
+    await approval_notification_hooks.handle(event, gl, *args, **kwargs)
+
+
+@bot.router.register("Merge Request Hook", action="unapproved")
+@bot.router.register("Merge Request Hook", action="unapproval")
+async def merge_request_unapproval_event(event, gl, *args, **kwargs):
+    await approval_notification_hooks.handle(event, gl, *args, **kwargs)
+
+
 @bot.router.register("Note Hook", noteable_type="MergeRequest")
 async def note_merge_request_event(event, gl, *args, **kwargs):
     if not ignore_event(event):

src/approval_notification.py

@@ -0,0 +1,204 @@
+# Copyright 2026 Lei Zhang
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import logging
+from abc import ABC, abstractmethod
+from dataclasses import asdict, dataclass
+from typing import Any, Dict, Mapping, Optional
+
+APPROVAL_ACTIONS = frozenset(("approval", "approved", "unapproval", "unapproved"))
+NORMALIZED_ACTIONS = {
+    "approval": "approval",
+    "approved": "approval",
+    "unapproval": "unapproval",
+    "unapproved": "unapproval",
+}
+
+
+@dataclass
+class MergeRequestNotification:
+    """Normalized, channel-independent information about an MR review action."""
+
+    source: str
+    event_type: str
+    action: str
+    webhook_action: str
+    message: str
+    project: Dict[str, Any]
+    merge_request: Dict[str, Any]
+    actor: Dict[str, Any]
+    occurred_at: Optional[str]
+    raw_payload: Optional[Mapping[str, Any]] = None
+
+
+class Channel(ABC):
+    """Asynchronous destination for normalized notifications."""
+
+    @abstractmethod
+    async def send(self, notification: MergeRequestNotification) -> None:
+        raise NotImplementedError
+
+
+class LogChannel(Channel):
+    """Write normalized notifications as searchable JSON log records."""
+
+    def __init__(self, logger: Optional[logging.Logger] = None):
+        self.logger = logger or logging.getLogger(__name__)
+
+    async def send(self, notification: MergeRequestNotification) -> None:
+        self.logger.info(json.dumps(asdict(notification), ensure_ascii=False))
+
+
+def _first_value(data: Mapping[str, Any], *keys: str) -> Any:
+    for key in keys:
+        value = data.get(key)
+        if value is not None and value != "":
+            return value
+    return None
+
+
+def _require_mapping(data: Any, name: str) -> Mapping[str, Any]:
+    if not isinstance(data, Mapping):
+        raise ValueError(f"{name} must be an object")
+    return data
+
+
+def _require_value(data: Mapping[str, Any], name: str) -> Any:
+    value = data.get(name)
+    if value is None or value == "":
+        raise ValueError(f"{name} is required")
+    return value
+
+
+def _build_message(
+    action: str,
+    actor_name: str,
+    title: str,
+    iid: Any,
+    project_label: str,
+    mr_url: Optional[str],
+) -> str:
+    if action == "approval":
+        action_text = "approved"
+    else:
+        action_text = "canceled approval (unapproval) for"
+
+    message = f"{actor_name} {action_text} MR !{iid}: {title} (project: {project_label})"
+    if mr_url:
+        message += f" {mr_url}"
+    else:
+        message += " (MR URL unavailable)"
+    return message
+
+
+def build_notification(data: Mapping[str, Any]) -> MergeRequestNotification:
+    """Convert a GitLab merge request webhook payload into a notification."""
+
+    payload = _require_mapping(data, "payload")
+    attributes = _require_mapping(payload.get("object_attributes"), "object_attributes")
+    project_data = _require_mapping(payload.get("project"), "project")
+    actor_data = _require_mapping(payload.get("user"), "user")
+
+    webhook_action = _require_value(attributes, "action")
+    if not isinstance(webhook_action, str) or webhook_action not in APPROVAL_ACTIONS:
+        raise ValueError(f"unsupported approval action: {webhook_action}")
+    action = NORMALIZED_ACTIONS[webhook_action]
+
+    project_id = _require_value(project_data, "id")
+    iid = _require_value(attributes, "iid")
+    username = _require_value(actor_data, "username")
+
+    project_path = _first_value(project_data, "path_with_namespace", "path", "name")
+    project_url = _first_value(project_data, "web_url", "url")
+    mr_url = _first_value(attributes, "url", "web_url")
+    if mr_url is None and project_url is not None:
+        mr_url = f"{str(project_url).rstrip('/')}/-/merge_requests/{iid}"
+
+    title = attributes.get("title")
+    title_text = str(title) if title is not None and title != "" else "(untitled)"
+    project_label = str(project_path) if project_path is not None else str(project_id)
+    actor_name = str(_first_value(actor_data, "name", "username"))
+    if actor_name != str(username):
+        actor_name = f"{actor_name} (@{username})"
+
+    return MergeRequestNotification(
+        source="gitlab",
+        event_type="merge_request_review",
+        action=action,
+        webhook_action=webhook_action,
+        message=_build_message(
+            action,
+            actor_name,
+            title_text,
+            iid,
+            project_label,
+            str(mr_url) if mr_url is not None else None,
+        ),
+        project={
+            "id": project_id,
+            "path": project_path,
+            "url": project_url,
+        },
+        merge_request={
+            "iid": iid,
+            "title": title,
+            "url": mr_url,
+        },
+        actor={
+            "id": actor_data.get("id"),
+            "username": username,
+            "name": actor_data.get("name"),
+        },
+        occurred_at=_first_value(attributes, "actioned_at", "updated_at"),
+    )
+
+
+class ApprovalNotificationHooks:
+    """Handle approval webhooks without invoking any GitLab API."""
+
+    def __init__(self, channel: Channel, logger: Optional[logging.Logger] = None):
+        self.channel = channel
+        self.logger = logger or logging.getLogger(__name__)
+
+    async def handle(self, event, *args, **kwargs) -> None:
+        try:
+            data = event.data
+            if not isinstance(data, Mapping):
+                raise ValueError("payload must be an object")
+            attributes = data.get("object_attributes")
+            if not isinstance(attributes, Mapping):
+                raise ValueError("object_attributes must be an object")
+            action = attributes.get("action")
+            if action is None or action == "":
+                raise ValueError("object_attributes.action is required")
+            if action not in APPROVAL_ACTIONS:
+                self.logger.debug("Skip approval notification for action=%r", action)
+                return
+            notification = build_notification(data)
+        except Exception as exc:
+            self.logger.error("invalid approval webhook: %s", exc)
+            return
+
+        try:
+            await self.channel.send(notification)
+        except Exception as exc:
+            self.logger.error(
+                "approval notification channel failed (action=%s, project=%s, mr_iid=%s): %s",
+                notification.action,
+                notification.project.get("path") or notification.project.get("id"),
+                notification.merge_request.get("iid"),
+                exc,
+                exc_info=True,
+            )

tests/conftest.py

@@ -0,0 +1,18 @@
+"""Keep test configuration independent from a developer's local .env file."""
+
+import os
+
+os.environ.update(
+    {
+        "BOT_GIT_COMMIT_MESSAGE_CHECK_ENABLED": "true",
+        "BOT_GIT_COMMIT_SUBJECT_REGEX_ENABLED": "true",
+        "BOT_GIT_COMMIT_SUBJECT_REGEX": (
+            r"^(\[(fix|feat)\]:\[.*]\[.*\]|"
+            r"\[(docs|style|ref|test|chore|tag|revert|perf)\]:\[.*\])$"
+        ),
+        "BOT_GIT_EMAIL_DOMAIN": "asiainfo.com",
+        "BOT_GITLAB_MERGE_REQUEST_MILESTONE_REQUIRED": "false",
+        "BOT_GITLAB_MERGE_REQUEST_ISSUE_REQUIRED": "false",
+        "BOT_GITLAB_MERGE_REQUEST_APPROVAL_ENABLED": "true",
+    }
+)

tests/fixtures/approval_webhook.py

@@ -0,0 +1,45 @@
+"""Representative approval and unapproval webhook payloads.
+
+These payloads follow the GitLab merge request webhook shape. Replace or
+augment them with a redacted GitLab Community Edition 16.11 capture when the
+target-instance smoke fixture is available.
+"""
+
+from copy import deepcopy
+
+
+def make_approval_webhook(action="approval", username="reviewer"):
+    return {
+        "object_kind": "merge_request",
+        "event_type": "merge_request",
+        "user": {
+            "id": 7,
+            "name": "Reviewer",
+            "username": username,
+        },
+        "project": {
+            "id": 76,
+            "name": "Project",
+            "path_with_namespace": "group/project",
+            "web_url": "https://gitlab.example.com/group/project",
+        },
+        "object_attributes": {
+            "action": action,
+            "iid": 12,
+            "title": "Add feature",
+            "url": "https://gitlab.example.com/group/project/-/merge_requests/12",
+            "updated_at": "2026-08-02T10:00:00Z",
+        },
+    }
+
+
+APPROVAL_WEBHOOK = make_approval_webhook()
+UNAPPROVAL_WEBHOOK = make_approval_webhook(action="unapproval")
+
+
+def copy_webhook(action="approval", username="reviewer"):
+    if action == "approval" and username == "reviewer":
+        return deepcopy(APPROVAL_WEBHOOK)
+    if action == "unapproval" and username == "reviewer":
+        return deepcopy(UNAPPROVAL_WEBHOOK)
+    return make_approval_webhook(action=action, username=username)