integrations: Add support for GitLab access token expiry events.

Fixes #34405.

zerver/webhooks/gitlab/fixtures/resource_access_token_hook__group_expiry.json

@@ -0,0 +1,17 @@
+{
+    "object_kind": "access_token",
+    "group": {
+      "group_name": "Twitter",
+      "group_path": "twitter",
+      "group_id": 35,
+      "full_path": "twitter"
+    },
+    "object_attributes": {
+      "user_id": 90,
+      "created_at": "2024-01-24 16:27:40 UTC",
+      "id": 25,
+      "name": "acd",
+      "expires_at": "2024-01-26"
+    },
+    "event_name": "expiring_access_token"
+  }

zerver/webhooks/gitlab/fixtures/resource_access_token_hook__project_expiry.json

@@ -0,0 +1,29 @@
+{
+    "object_kind": "access_token",
+    "project": {
+      "id": 7,
+      "name": "Flight",
+      "description": "Eum dolore maxime atque reprehenderit voluptatem.",
+      "web_url": "https://example.com/flightjs/Flight",
+      "avatar_url": null,
+      "git_ssh_url": "ssh://git@example.com/flightjs/Flight.git",
+      "git_http_url": "https://example.com/flightjs/Flight.git",
+      "namespace": "Flightjs",
+      "visibility_level": 0,
+      "path_with_namespace": "flightjs/Flight",
+      "default_branch": "master",
+      "ci_config_path": null,
+      "homepage": "https://example.com/flightjs/Flight",
+      "url": "ssh://git@example.com/flightjs/Flight.git",
+      "ssh_url": "ssh://git@example.com/flightjs/Flight.git",
+      "http_url": "https://example.com/flightjs/Flight.git"
+    },
+    "object_attributes": {
+      "user_id": 90,
+      "created_at": "2024-01-24 16:27:40 UTC",
+      "id": 25,
+      "name": "acd",
+      "expires_at": "2024-01-26"
+    },
+    "event_name": "expiring_access_token"
+  }

zerver/webhooks/gitlab/tests.py

@@ -697,3 +697,19 @@ A trivial change that should probably be ignored.
         expected_message = "kolanuvarun739 deactivated the feature flag [sample-feature-flag](https://gitlab.com/kolanuvarun/sample/-/feature_flags)."
 
         self.check_webhook("feature_flag_hook__deactivated", expected_topic_name, expected_message)
+
+    def test_resource_access_token_project_expiry(self) -> None:
+        expected_topic_name = "Flight"
+        expected_message = "The access token [acd](https://example.com/flightjs/Flight/-/settings/access_tokens) will expire on Jan 26, 2024."
+
+        self.check_webhook(
+            "resource_access_token_hook__project_expiry", expected_topic_name, expected_message
+        )
+
+    def test_resource_access_token_group_expiry(self) -> None:
+        expected_topic_name = "Twitter"
+        expected_message = "The access token [acd](https://gitlab.com/groups/twitter/-/settings/access_tokens) will expire on Jan 26, 2024."
+
+        self.check_webhook(
+            "resource_access_token_hook__group_expiry", expected_topic_name, expected_message
+        )

zerver/webhooks/gitlab/view.py

@@ -1,4 +1,5 @@
 import re
+from datetime import datetime, timezone
 from typing import Protocol
 
 from django.http import HttpRequest, HttpResponse
@@ -38,6 +39,8 @@ DESIGN_COMMENT_MESSAGE_TEMPLATE = (
 
 FEATURE_FLAG_MESSAGE_TEMPLATE = "{user} {action} the feature flag [{name}]({url})."
 
+ACCESS_TOKEN_EXPIRY_MESSAGE_TEMPLATE = "The access token [{name}]({url}) will expire on {date}."
+
 
 def fixture_to_headers(fixture_name: str) -> dict[str, str]:
     if fixture_name.startswith("build"):
@@ -423,6 +426,35 @@ def get_feature_flag_event_body(payload: WildValue, include_title: bool) -> str:
     )
 
 
+def get_access_token_page_url(payload: WildValue) -> str:
+    """
+    Generate the URL for the access tokens based on whether it's
+    for a group or a project.
+    """
+    if "group" in payload:
+        group_path = payload["group"]["group_path"].tame(check_string)
+        return f"https://gitlab.com/groups/{group_path}/-/settings/access_tokens"
+
+    project_url = payload["project"]["web_url"].tame(check_string)
+    return f"{project_url}/-/settings/access_tokens"
+
+
+def get_resource_access_token_expiry_event_body(payload: WildValue, include_title: bool) -> str:
+    access_token = payload["object_attributes"]
+    expiry_date = access_token["expires_at"].tame(check_string)
+    formatted_date = (
+        datetime.strptime(expiry_date, "%Y-%m-%d")
+        .replace(tzinfo=timezone.utc)
+        .strftime("%b %d, %Y")
+    )
+
+    return ACCESS_TOKEN_EXPIRY_MESSAGE_TEMPLATE.format(
+        name=access_token["name"].tame(check_string),
+        url=get_access_token_page_url(payload),
+        date=formatted_date,
+    )
+
+
 def get_repo_name(payload: WildValue) -> str:
     if "project" in payload:
         return payload["project"]["name"].tame(check_string)
@@ -511,6 +543,7 @@ EVENT_FUNCTION_MAPPER: dict[str, EventFunction] = {
     "Pipeline Hook": get_pipeline_event_body,
     "Release Hook": get_release_event_body,
     "Feature Flag Hook": get_feature_flag_event_body,
+    "Resource Access Token Hook": get_resource_access_token_expiry_event_body,
 }
 
 ALL_EVENT_TYPES = list(EVENT_FUNCTION_MAPPER.keys())
@@ -619,6 +652,9 @@ def get_topic_based_on_event(event: str, payload: WildValue, use_merge_request_t
             id=payload["snippet"]["id"].tame(check_int),
             title=payload["snippet"]["title"].tame(check_string),
         )
+
+    elif event == "Resource Access Token Hook" and payload.get("group"):
+        return payload["group"]["group_name"].tame(check_string)
     return get_repo_name(payload)