From 044edad65c3ef9a9f431a6460829a83fb1ab4b45 Mon Sep 17 00:00:00 2001 From: zhanglei Date: Sun, 2 Aug 2026 22:05:12 +0800 Subject: [PATCH 1/2] feat: add pipeline status notifications --- README.md | 4 +- README_ZH.md | 4 +- gitlab_bot.py | 7 + src/channels/base.py | 4 +- src/channels/log.py | 4 +- src/hooks/pipeline_notification.py | 239 +++++++++++++++++++ src/notifications/__init__.py | 4 +- src/notifications/model.py | 24 +- tests/fixtures/pipeline_webhook.py | 60 +++++ tests/test_hook_modules.py | 3 + tests/test_notification_contracts.py | 3 +- tests/test_notification_module_boundaries.py | 13 + tests/test_pipeline_notification.py | 119 +++++++++ 13 files changed, 478 insertions(+), 10 deletions(-) create mode 100644 src/hooks/pipeline_notification.py create mode 100644 tests/fixtures/pipeline_webhook.py create mode 100644 tests/test_pipeline_notification.py diff --git a/README.md b/README.md index 9617ddc..3d68a9a 100644 --- a/README.md +++ b/README.md @@ -52,7 +52,8 @@ This is a GitLab(13.2+) bot that utilizes [webhooks](https://docs.gitlab.com/ee/ * Merge Request Events * Wiki Page Events (Not yet) * Tag Events (Not yet) -* Pipeline Events (Not yet) +* Pipeline Events + * Pipeline success/failure notifications * Build Events (Not yet) ## Development @@ -123,6 +124,7 @@ To enable the GitLab Bot to respond to events, you need to configure a Webhook i - `Comments` (comment events) - `Issues events` (issue events) - `Merge request events` (merge request events) + - `Pipeline events` (Pipeline success/failure notifications) 5. **Save the Webhook**: - Click on the `Add webhook` button to save your configuration. diff --git a/README_ZH.md b/README_ZH.md index ab24343..ef0fe5f 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -52,7 +52,8 @@ * 合并请求事件 * 维基页面事件（尚未支持） * 标签事件（尚未支持） -* 管道事件（尚未支持） +* 管道事件 + * Pipeline 成功/失败通知 * 构建事件（尚未支持） ## 本地开发 @@ -121,6 +122,7 @@ coolbeevip/gitlab-bot - `评论`（评论事件） - `问题事件`（问题事件） - `合并请求事件`（合并请求事件） + - `管道事件`（Pipeline 成功/失败通知） 5. **保存 Webhook**： - 点击 `添加 webhook` 按钮以保存您的配置。 diff --git a/gitlab_bot.py b/gitlab_bot.py index e195f14..116f4bb 100644 --- a/gitlab_bot.py +++ b/gitlab_bot.py @@ -35,6 +35,7 @@ from src.hooks.merge_notification import MergeRequestNotificationHooks from src.hooks.merge_request import MergeRequestHooks from src.hooks.note import NoteHooks +from src.hooks.pipeline_notification import PipelineNotificationHooks from src.logs import print_event load_dotenv() # isort:skip @@ -67,6 +68,7 @@ def _load_gitlab_bot(): note_hooks = NoteHooks() notification_channel = LogChannel() approval_notification_hooks = ApprovalNotificationHooks(notification_channel) +pipeline_notification_hooks = PipelineNotificationHooks(notification_channel) notification_delivery_store = NotificationDeliveryStore( merge_notification_db_path, sending_timeout_seconds=merge_notification_sending_timeout_seconds, @@ -145,6 +147,11 @@ async def merge_request_merged_event(event, gl, *args, **kwargs): await merge_request_notification_hooks.handle(event, gl, *args, **kwargs) +@bot.router.register("Pipeline Hook") +async def pipeline_event(event, gl, *args, **kwargs): + await pipeline_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): diff --git a/src/channels/base.py b/src/channels/base.py index 574d082..4d73c83 100644 --- a/src/channels/base.py +++ b/src/channels/base.py @@ -14,12 +14,12 @@ from abc import ABC, abstractmethod -from ..notifications.model import MergeRequestNotification +from ..notifications.model import Notification class Channel(ABC): """Asynchronous destination for normalized notifications.""" @abstractmethod - async def send(self, notification: MergeRequestNotification) -> None: + async def send(self, notification: Notification) -> None: raise NotImplementedError diff --git a/src/channels/log.py b/src/channels/log.py index bec1d7d..0d6b720 100644 --- a/src/channels/log.py +++ b/src/channels/log.py @@ -17,7 +17,7 @@ from dataclasses import asdict from typing import Optional -from ..notifications.model import MergeRequestNotification +from ..notifications.model import Notification from .base import Channel @@ -27,5 +27,5 @@ class LogChannel(Channel): def __init__(self, logger: Optional[logging.Logger] = None): self.logger = logger or logging.getLogger(__name__) - async def send(self, notification: MergeRequestNotification) -> None: + async def send(self, notification: Notification) -> None: self.logger.info(json.dumps(asdict(notification), ensure_ascii=False)) diff --git a/src/hooks/pipeline_notification.py b/src/hooks/pipeline_notification.py new file mode 100644 index 0000000..773655c --- /dev/null +++ b/src/hooks/pipeline_notification.py @@ -0,0 +1,239 @@ +# 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 logging +from typing import Any, Dict, Mapping, Optional + +from ..channels.base import Channel +from ..notifications.model import PipelineNotification + +PIPELINE_STATUSES = frozenset(("success", "failed")) +PIPELINE_NOTIFICATION_EVENT_TYPE = "pipeline_lifecycle" +PIPELINE_WEBHOOK_ACTION = "status_changed" +STATUS_LABELS = { + "success": "成功", + "failed": "失败", +} + +__all__ = [ + "PIPELINE_STATUSES", + "PIPELINE_NOTIFICATION_EVENT_TYPE", + "PIPELINE_WEBHOOK_ACTION", + "PipelineNotification", + "Channel", + "build_pipeline_notification", + "PipelineNotificationHooks", +] + + +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 _normalize_actor(data: Any) -> Dict[str, Any]: + if not isinstance(data, Mapping): + return {"id": None, "username": None, "name": None} + username = _first_value(data, "username", "user_name") + name = _first_value(data, "name", "user_name") or username + return { + "id": data.get("id"), + "username": username, + "name": name, + } + + +def _normalize_merge_request(data: Any) -> Optional[Dict[str, Any]]: + if not isinstance(data, Mapping): + return None + merge_request = { + "iid": _first_value(data, "iid"), + "title": _first_value(data, "title"), + "url": _first_value(data, "url", "web_url"), + "source_branch": _first_value(data, "source_branch"), + "target_branch": _first_value(data, "target_branch"), + } + if not any(value is not None for value in merge_request.values()): + return None + return merge_request + + +def _pipeline_idempotency_key( + payload: Mapping[str, Any], + attributes: Mapping[str, Any], + project_id: Any, + pipeline_id: Any, + status: str, +) -> str: + event_marker = _first_value(payload, "webhook_id", "idempotency_key") + if event_marker is None: + event_marker = _first_value(attributes, "webhook_id", "idempotency_key") + if event_marker is not None: + return f"gitlab:pipeline:{project_id}:{pipeline_id}:{status}:{event_marker}" + return f"gitlab:pipeline:{project_id}:{pipeline_id}:{status}" + + +def _event_idempotency_key(event) -> Optional[str]: + for attribute in ("webhook_id", "idempotency_key"): + value = getattr(event, attribute, None) + if value is not None and value != "": + return f"gitlab:pipeline:header:{value}" + + headers = getattr(event, "headers", None) + if isinstance(headers, Mapping): + normalized_headers = {str(key).lower(): value for key, value in headers.items()} + for header in ("webhook-id", "idempotency-key", "x-gitlab-webhook-id"): + value = normalized_headers.get(header) + if value is not None and value != "": + return f"gitlab:pipeline:header:{value}" + return None + + +def build_pipeline_notification(data: Mapping[str, Any]) -> PipelineNotification: + """Convert a GitLab Pipeline Hook 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") + + status = _require_value(attributes, "status") + if not isinstance(status, str) or status not in PIPELINE_STATUSES: + raise ValueError(f"unsupported pipeline status: {status}") + + project_id = _require_value(project_data, "id") + pipeline_id = _require_value(attributes, "id") + pipeline_iid = _first_value(attributes, "iid") + project_path = _first_value(project_data, "path_with_namespace", "path", "name") + project_url = _first_value(project_data, "web_url", "url") + pipeline_url = _first_value(attributes, "url", "web_url") + if pipeline_url is None and project_url is not None: + pipeline_url = f"{str(project_url).rstrip('/')}/-/pipelines/{pipeline_id}" + + merge_request = _normalize_merge_request(payload.get("merge_request")) + actor = _normalize_actor(payload.get("user")) + actor_name = actor.get("name") or actor.get("username") + project_label = str(project_path) if project_path is not None else str(project_id) + pipeline_label = f"Pipeline #{pipeline_id}" + if pipeline_iid is not None and str(pipeline_iid) != str(pipeline_id): + pipeline_label += f" (IID {pipeline_iid})" + + merge_request_label = "" + if merge_request is not None: + merge_request_iid = merge_request.get("iid") + merge_request_title = merge_request.get("title") or "标题不可用" + if merge_request_iid is not None: + merge_request_label = f" for MR !{merge_request_iid}「{merge_request_title}」" + + status_label = STATUS_LABELS[status] + details = [ + f"project: {project_label}", + f"ref: {_first_value(attributes, 'ref') or 'ref unavailable'}", + ] + if actor_name: + details.append(f"triggered by: {actor_name}") + if attributes.get("duration") is not None: + details.append(f"duration: {attributes['duration']}s") + if pipeline_url: + details.append(str(pipeline_url)) + else: + details.append("(Pipeline URL unavailable)") + + return PipelineNotification( + source="gitlab", + event_type=PIPELINE_NOTIFICATION_EVENT_TYPE, + action=status, + webhook_action=PIPELINE_WEBHOOK_ACTION, + status=status, + message=f"{pipeline_label}{merge_request_label} {status_label} ({status}): " + ", ".join(details), + project={ + "id": project_id, + "path": project_path, + "url": project_url, + }, + pipeline={ + "id": pipeline_id, + "iid": pipeline_iid, + "name": attributes.get("name"), + "status": status, + "detailed_status": attributes.get("detailed_status"), + "ref": attributes.get("ref"), + "sha": attributes.get("sha"), + "source": attributes.get("source"), + "url": pipeline_url, + "duration": attributes.get("duration"), + "queued_duration": attributes.get("queued_duration"), + "created_at": attributes.get("created_at"), + "finished_at": attributes.get("finished_at"), + }, + actor=actor, + occurred_at=_first_value(attributes, "finished_at", "updated_at", "created_at"), + merge_request=merge_request, + idempotency_key=_pipeline_idempotency_key(payload, attributes, project_id, pipeline_id, status), + ) + + +class PipelineNotificationHooks: + """Handle successful and failed GitLab Pipeline webhooks.""" + + 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") + status = attributes.get("status") + if status not in PIPELINE_STATUSES: + self.logger.debug("Skip pipeline notification for status=%r", status) + return + notification = build_pipeline_notification(data) + event_key = _event_idempotency_key(event) + if event_key is not None: + notification.idempotency_key = event_key + except Exception as exc: + self.logger.error("invalid pipeline webhook: %s", exc) + return + + try: + await self.channel.send(notification) + except Exception as exc: + self.logger.error( + "pipeline notification channel failed (status=%s, project=%s, pipeline_id=%s): %s", + notification.status, + notification.project.get("path") or notification.project.get("id"), + notification.pipeline.get("id"), + exc, + exc_info=True, + ) diff --git a/src/notifications/__init__.py b/src/notifications/__init__.py index 31103f9..21f2ee6 100644 --- a/src/notifications/__init__.py +++ b/src/notifications/__init__.py @@ -12,6 +12,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -from .model import MergeRequestNotification +from .model import MergeRequestNotification, Notification, PipelineNotification -__all__ = ["MergeRequestNotification"] +__all__ = ["MergeRequestNotification", "PipelineNotification", "Notification"] diff --git a/src/notifications/model.py b/src/notifications/model.py index 9a7175c..22a9f3a 100644 --- a/src/notifications/model.py +++ b/src/notifications/model.py @@ -13,7 +13,7 @@ # limitations under the License. from dataclasses import dataclass -from typing import Any, Dict, Mapping, Optional +from typing import Any, Dict, Mapping, Optional, Union @dataclass @@ -32,3 +32,25 @@ class MergeRequestNotification: raw_payload: Optional[Mapping[str, Any]] = None triggered_by: Optional[Dict[str, Any]] = None idempotency_key: Optional[str] = None + + +@dataclass +class PipelineNotification: + """Normalized, channel-independent information about a pipeline status.""" + + source: str + event_type: str + action: str + webhook_action: str + status: str + message: str + project: Dict[str, Any] + pipeline: Dict[str, Any] + actor: Dict[str, Any] + occurred_at: Optional[str] + merge_request: Optional[Dict[str, Any]] = None + raw_payload: Optional[Mapping[str, Any]] = None + idempotency_key: Optional[str] = None + + +Notification = Union[MergeRequestNotification, PipelineNotification] diff --git a/tests/fixtures/pipeline_webhook.py b/tests/fixtures/pipeline_webhook.py new file mode 100644 index 0000000..33f51f7 --- /dev/null +++ b/tests/fixtures/pipeline_webhook.py @@ -0,0 +1,60 @@ +"""Representative GitLab Pipeline Hook payloads.""" + +from copy import deepcopy + + +def make_pipeline_webhook(*, status="success", include_merge_request=True, username="pipeline-trigger"): + payload = { + "object_kind": "pipeline", + "user": { + "id": 7, + "name": "Pipeline Trigger", + "username": username, + }, + "project": { + "id": 76, + "name": "Project", + "path_with_namespace": "group/project", + "web_url": "https://gitlab.example.com/group/project", + }, + "object_attributes": { + "id": 31, + "iid": 3, + "name": "Pipeline for branch: master", + "ref": "master", + "tag": False, + "sha": "bcbb5ec396a2c0f828686f14fac9b80b780504f2", + "source": "push", + "status": status, + "detailed_status": "passed" if status == "success" else "failed", + "created_at": "2026-08-02T10:05:00Z", + "finished_at": "2026-08-02T10:06:03Z", + "duration": 63, + "queued_duration": 10, + "url": "https://gitlab.example.com/group/project/-/pipelines/31", + }, + "commit": { + "id": "bcbb5ec396a2c0f828686f14fac9b80b780504f2", + "message": "Merge branch feature", + }, + } + if include_merge_request: + payload["merge_request"] = { + "id": 1, + "iid": 12, + "title": "Add feature", + "source_branch": "feature", + "target_branch": "master", + "url": "https://gitlab.example.com/group/project/-/merge_requests/12", + } + return payload + + +PIPELINE_WEBHOOK = make_pipeline_webhook() +FAILED_PIPELINE_WEBHOOK = make_pipeline_webhook(status="failed") + + +def copy_pipeline_webhook(**kwargs): + if not kwargs: + return deepcopy(PIPELINE_WEBHOOK) + return make_pipeline_webhook(**kwargs) diff --git a/tests/test_hook_modules.py b/tests/test_hook_modules.py index dbf425e..f9877bb 100644 --- a/tests/test_hook_modules.py +++ b/tests/test_hook_modules.py @@ -1,9 +1,12 @@ from src.hooks.approval_notification import ApprovalNotificationHooks, build_notification from src.hooks.merge_notification import MergeRequestNotificationHooks, build_merged_notification +from src.hooks.pipeline_notification import PipelineNotificationHooks, build_pipeline_notification def test_notification_hooks_are_available_from_responsibility_modules(): assert ApprovalNotificationHooks.__module__ == "src.hooks.approval_notification" assert MergeRequestNotificationHooks.__module__ == "src.hooks.merge_notification" + assert PipelineNotificationHooks.__module__ == "src.hooks.pipeline_notification" assert build_notification.__module__ == "src.hooks.approval_notification" assert build_merged_notification.__module__ == "src.hooks.merge_notification" + assert build_pipeline_notification.__module__ == "src.hooks.pipeline_notification" diff --git a/tests/test_notification_contracts.py b/tests/test_notification_contracts.py index c9e8360..a670057 100644 --- a/tests/test_notification_contracts.py +++ b/tests/test_notification_contracts.py @@ -3,13 +3,14 @@ from src.delivery.coordinator import NotificationDelivery from src.delivery.idempotent_channel import DurableIdempotentChannel from src.delivery.sqlite import NotificationDeliveryStore -from src.notifications.model import MergeRequestNotification +from src.notifications.model import MergeRequestNotification, PipelineNotification def test_notification_contracts_are_available_from_responsibility_modules(): assert Channel.__module__ == "src.channels.base" assert LogChannel.__module__ == "src.channels.log" assert MergeRequestNotification.__module__ == "src.notifications.model" + assert PipelineNotification.__module__ == "src.notifications.model" def test_delivery_components_are_available_from_responsibility_modules(): diff --git a/tests/test_notification_module_boundaries.py b/tests/test_notification_module_boundaries.py index d1e372a..f5aefad 100644 --- a/tests/test_notification_module_boundaries.py +++ b/tests/test_notification_module_boundaries.py @@ -7,8 +7,10 @@ import gitlab_bot from src.hooks.approval_notification import ApprovalNotificationHooks from src.hooks.merge_notification import MergeRequestNotificationHooks +from src.hooks.pipeline_notification import PipelineNotificationHooks from tests.fixtures.approval_webhook import copy_webhook from tests.fixtures.merge_webhook import copy_merge_webhook +from tests.fixtures.pipeline_webhook import copy_pipeline_webhook PROJECT_ROOT = Path(__file__).resolve().parents[1] SRC_ROOT = PROJECT_ROOT / "src" @@ -34,6 +36,7 @@ def test_notification_hooks_only_depend_on_notification_model_and_channel_contra notification_hook_paths = ( SRC_ROOT / "hooks" / "approval_notification.py", SRC_ROOT / "hooks" / "merge_notification.py", + SRC_ROOT / "hooks" / "pipeline_notification.py", ) for path in notification_hook_paths: imports = tuple(_import_modules(path)) @@ -102,3 +105,13 @@ async def send(self, notification): ) ) assert len(merge_channel.notifications) == 1 + + pipeline_channel = RecordingChannel() + monkeypatch.setattr(gitlab_bot, "pipeline_notification_hooks", PipelineNotificationHooks(pipeline_channel)) + asyncio.run( + gitlab_bot.bot.router.dispatch( + Event(copy_pipeline_webhook(status="success"), event="Pipeline Hook"), + None, + ) + ) + assert len(pipeline_channel.notifications) == 1 diff --git a/tests/test_pipeline_notification.py b/tests/test_pipeline_notification.py new file mode 100644 index 0000000..3ec3d19 --- /dev/null +++ b/tests/test_pipeline_notification.py @@ -0,0 +1,119 @@ +import asyncio +import json +import logging +from types import SimpleNamespace + +import pytest +from gidgetlab.sansio import Event + +import gitlab_bot +from src.channels.log import LogChannel +from src.hooks.pipeline_notification import PipelineNotificationHooks, build_pipeline_notification +from tests.fixtures.pipeline_webhook import copy_pipeline_webhook + + +class RecordingChannel: + def __init__(self): + self.notifications = [] + + async def send(self, notification): + self.notifications.append(notification) + + +def make_event(payload=None): + return SimpleNamespace(data=payload or copy_pipeline_webhook()) + + +@pytest.mark.parametrize("status", ["success", "failed"]) +def test_build_pipeline_notification_contains_pipeline_and_merge_request_details(status): + notification = build_pipeline_notification(copy_pipeline_webhook(status=status)) + + assert notification.source == "gitlab" + assert notification.event_type == "pipeline_lifecycle" + assert notification.action == status + assert notification.status == status + assert notification.webhook_action == "status_changed" + assert notification.project["id"] == 76 + assert notification.pipeline["id"] == 31 + assert notification.pipeline["status"] == status + assert notification.merge_request["iid"] == 12 + assert "Pipeline #31" in notification.message + assert "MR !12" in notification.message + assert status in notification.message + assert "group/project" in notification.message + assert "https://gitlab.example.com/group/project/-/pipelines/31" in notification.message + + +def test_build_branch_pipeline_without_merge_request(): + notification = build_pipeline_notification(copy_pipeline_webhook(include_merge_request=False)) + + assert notification.merge_request is None + assert "MR !" not in notification.message + + +@pytest.mark.parametrize("status", ["created", "pending", "running", "canceled", "skipped"]) +def test_non_terminal_pipeline_statuses_are_skipped(status): + channel = RecordingChannel() + + asyncio.run(PipelineNotificationHooks(channel).handle(make_event(copy_pipeline_webhook(status=status)))) + + assert channel.notifications == [] + + +@pytest.mark.parametrize("status", ["success", "failed"]) +def test_pipeline_route_sends_success_and_failure_notifications(monkeypatch, status): + channel = RecordingChannel() + monkeypatch.setattr(gitlab_bot, "pipeline_notification_hooks", PipelineNotificationHooks(channel)) + event = Event(copy_pipeline_webhook(status=status), event="Pipeline Hook") + + asyncio.run(gitlab_bot.bot.router.dispatch(event, None)) + + assert len(channel.notifications) == 1 + assert channel.notifications[0].status == status + + +def test_pipeline_notification_uses_webhook_id_when_available(): + channel = RecordingChannel() + event = SimpleNamespace(data=copy_pipeline_webhook(), headers={"webhook-id": "pipeline-hook-123"}) + + asyncio.run(PipelineNotificationHooks(channel).handle(event)) + + assert channel.notifications[0].idempotency_key == "gitlab:pipeline:header:pipeline-hook-123" + + +def test_log_channel_emits_structured_pipeline_json(caplog): + logger = logging.getLogger("test.pipeline_notification") + channel = LogChannel(logger) + notification = build_pipeline_notification(copy_pipeline_webhook(status="failed")) + caplog.set_level(logging.INFO, logger=logger.name) + + asyncio.run(channel.send(notification)) + + payload = json.loads(caplog.records[-1].message) + assert payload["status"] == "failed" + assert payload["pipeline"]["id"] == 31 + assert payload["merge_request"]["iid"] == 12 + + +def test_invalid_pipeline_payload_is_logged_and_skipped(caplog): + logger = logging.getLogger("test.pipeline_notification.invalid") + channel = RecordingChannel() + hooks = PipelineNotificationHooks(channel, logger=logger) + caplog.set_level(logging.ERROR, logger=logger.name) + + asyncio.run(hooks.handle(SimpleNamespace(data={"object_attributes": {"status": "success"}}))) + + assert channel.notifications == [] + assert "invalid pipeline webhook" in caplog.text + + +def test_pipeline_notification_does_not_call_gitlab_api(): + channel = RecordingChannel() + api = SimpleNamespace( + getitem=lambda *_args, **_kwargs: pytest.fail("pipeline notification must not call GitLab API"), + post=lambda *_args, **_kwargs: pytest.fail("pipeline notification must not call GitLab API"), + ) + + asyncio.run(PipelineNotificationHooks(channel).handle(make_event(), api)) + + assert len(channel.notifications) == 1 From dc4d0301d3062487ffe9a42f2e50c806a6cf1eb8 Mon Sep 17 00:00:00 2001 From: zhanglei Date: Sun, 2 Aug 2026 23:13:34 +0800 Subject: [PATCH 2/2] feat: add Feishu notification channel --- .env.example | 16 ++ README.md | 26 +++ README_ZH.md | 28 ++- gitlab_bot.py | 46 +++- pyproject.toml | 1 + src/channels/__init__.py | 10 +- src/channels/dispatcher.py | 71 ++++++ src/channels/feishu.py | 319 ++++++++++++++++++++++++++ src/config.py | 10 + src/delivery/coordinator.py | 16 +- src/delivery/idempotent_channel.py | 29 ++- src/delivery/sqlite.py | 199 +++++++++++++--- src/hooks/pipeline_notification.py | 17 +- tests/test_feishu_channel.py | 175 ++++++++++++++ tests/test_feishu_delivery.py | 141 ++++++++++++ tests/test_notification_delivery.py | 33 +++ tests/test_notification_dispatcher.py | 42 ++++ uv.lock | 2 + 18 files changed, 1127 insertions(+), 54 deletions(-) create mode 100644 .env.example create mode 100644 src/channels/dispatcher.py create mode 100644 src/channels/feishu.py create mode 100644 tests/test_feishu_channel.py create mode 100644 tests/test_feishu_delivery.py create mode 100644 tests/test_notification_dispatcher.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..48f0960 --- /dev/null +++ b/.env.example @@ -0,0 +1,16 @@ +# GitLab Bot configuration +BOT_GITLAB_USERNAME= +BOT_GITLAB_URL= +BOT_GITLAB_TOKEN= + +# Feishu notifications are disabled by default. +FEISHU_ENABLED=false +FEISHU_APP_ID= +FEISHU_APP_SECRET= +FEISHU_CHAT_ID= +FEISHU_BOT_OPEN_ID= +FEISHU_REQUEST_TIMEOUT_SECONDS=10 + +# Durable notification recovery +MERGE_NOTIFICATION_MAX_ATTEMPTS=5 +MERGE_NOTIFICATION_RETRY_BACKOFF_SECONDS=1 diff --git a/README.md b/README.md index 3d68a9a..0977888 100644 --- a/README.md +++ b/README.md @@ -132,6 +132,28 @@ To enable the GitLab Bot to respond to events, you need to configure a Webhook i 6. **Test the Webhook (Optional)**: - To confirm if the Webhook is set up correctly, locate the newly added Webhook in the Webhooks list and use the `Test` button to send a test request. +## Feishu Notifications + +When enabled, the bot sends normalized MR approval, MR merge, and Pipeline success/failure notifications to one Feishu group while keeping the structured log channel. Feishu is disabled by default. + +Before enabling it, create a Feishu custom app, grant permission to send group messages, add the app to the target group, and confirm the runtime can reach `https://open.feishu.cn` over HTTPS. + +```shell +FEISHU_ENABLED=true +FEISHU_APP_ID= +FEISHU_APP_SECRET= +FEISHU_CHAT_ID= +# Optional: +FEISHU_BOT_OPEN_ID= +FEISHU_REQUEST_TIMEOUT_SECONDS=10 +``` + +`FEISHU_APP_SECRET` must be injected as a secret and never committed or printed. The MVP sends text messages to one `FEISHU_CHAT_ID`; cards, rich text, multiple groups, and Feishu callbacks are not supported. + +MR merge and Pipeline Feishu deliveries are persisted with per-target idempotency and can be recovered or replayed after failures. Approval/unapproval keeps the existing direct-send failure behavior. To roll back, set `FEISHU_ENABLED=false`; existing log and GitLab processing remain available. + +Recovery can be tuned with `MERGE_NOTIFICATION_MAX_ATTEMPTS` (default `5`) and `MERGE_NOTIFICATION_RETRY_BACKOFF_SECONDS` (default `1`). Automatic retries are bounded; failed records remain available for manual replay. + ## Environment Variables **`BOT_GITLAB_USERNAME` / `BOT_GITLAB_URL` / `BOT_GITLAB_TOKEN`** @@ -150,6 +172,10 @@ Supports both Chinese (zh) and English (en) languages by default. These variables specify the host and port on which the bot will run. By default, the bot will run on the IP address 0.0.0.0 and port number 9998. +**`FEISHU_ENABLED` / `FEISHU_APP_ID` / `FEISHU_APP_SECRET` / `FEISHU_CHAT_ID` / `FEISHU_BOT_OPEN_ID` / `FEISHU_REQUEST_TIMEOUT_SECONDS`** + +These variables configure the optional Feishu notification channel. `FEISHU_ENABLED` defaults to `false`; the other required values are checked when it is enabled. Keep the App Secret outside source control and logs. See [Feishu Notifications](#feishu-notifications) for permissions, rollback, and delivery recovery behavior. + **`BOT_GIT_EMAIL_DOMAIN`** This configuration specifies the email domain that will be used for email addresses when making Git commits. For example: diff --git a/README_ZH.md b/README_ZH.md index ef0fe5f..30b3902 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -129,7 +129,29 @@ coolbeevip/gitlab-bot 6. **测试 Webhook（可选）**： - 为确认 Webhook 是否设置正确，在 Webhooks 列表中找到新添加的 Webhook，并使用 `测试` 按钮发送测试请求。 - + +## 飞书通知 + +启用后，Bot 会在保留结构化日志的同时，将 MR 审批、MR 合并和 Pipeline 成功/失败通知发送到一个飞书群。飞书默认关闭。 + +启用前请创建飞书自建应用，授予发送群消息权限，将应用加入目标群，并确认运行环境可以通过 HTTPS 访问 `https://open.feishu.cn`。 + +```shell +FEISHU_ENABLED=true +FEISHU_APP_ID=<飞书应用 App ID> +FEISHU_APP_SECRET=<飞书应用 Secret> +FEISHU_CHAT_ID=<目标群 Chat ID> +# 可选： +FEISHU_BOT_OPEN_ID=<需要被 @ 的用户 Open ID> +FEISHU_REQUEST_TIMEOUT_SECONDS=10 +``` + +`FEISHU_APP_SECRET` 必须通过 Secret 注入，不能提交到仓库或打印到日志。MVP 只向一个 `FEISHU_CHAT_ID` 发送文本消息，不支持卡片、富文本、多群路由和飞书回调。 + +MR 合并和 Pipeline 的飞书投递会按目标持久化幂等状态，失败后支持恢复或补发；审批/取消审批沿用现有直接发送失败记录行为。回滚时设置 `FEISHU_ENABLED=false`，日志和 GitLab 处理仍保持可用。 + +可通过 `MERGE_NOTIFICATION_MAX_ATTEMPTS`（默认 `5`）和 `MERGE_NOTIFICATION_RETRY_BACKOFF_SECONDS`（默认 `1`）调整恢复策略。自动重试次数有限，失败记录仍可人工补发。 + ## 环境变量 **`BOT_GITLAB_USERNAME` / `BOT_GITLAB_URL` / `BOT_GITLAB_TOKEN`** @@ -148,6 +170,10 @@ coolbeevip/gitlab-bot 这些变量指定机器人运行的主机和端口。默认情况下，机器人将在 IP 地址 0.0.0.0 和端口号 9998 上运行。 +**`FEISHU_ENABLED` / `FEISHU_APP_ID` / `FEISHU_APP_SECRET` / `FEISHU_CHAT_ID` / `FEISHU_BOT_OPEN_ID` / `FEISHU_REQUEST_TIMEOUT_SECONDS`** + +这些变量用于配置可选的飞书通知 Channel。`FEISHU_ENABLED` 默认是 `false`；启用后会校验其他必填配置。请将 App Secret 保存在 Secret 管理系统中，不要提交到仓库或写入日志。权限、回滚和投递恢复说明见[飞书通知](#飞书通知)。 + **`BOT_GIT_EMAIL_DOMAIN`** 该配置指定在进行 Git 提交时使用的电子邮件域名。例如： diff --git a/gitlab_bot.py b/gitlab_bot.py index 116f4bb..1d28781 100644 --- a/gitlab_bot.py +++ b/gitlab_bot.py @@ -17,6 +17,8 @@ from dotenv import load_dotenv +from src.channels.dispatcher import NotificationDispatcher +from src.channels.feishu import FeishuChannel from src.channels.log import LogChannel from src.config import ( bot_gitlab_token, @@ -24,7 +26,15 @@ bot_gitlab_username, bot_host, bot_port, + feishu_app_id, + feishu_app_secret, + feishu_bot_open_id, + feishu_chat_id, + feishu_enabled, + feishu_request_timeout_seconds, merge_notification_db_path, + merge_notification_max_attempts, + merge_notification_retry_backoff_seconds, merge_notification_sending_timeout_seconds, ) from src.delivery.coordinator import NotificationDelivery @@ -66,23 +76,51 @@ def _load_gitlab_bot(): issue_hooks = IssueHooks() merge_request_hooks = MergeRequestHooks() note_hooks = NoteHooks() -notification_channel = LogChannel() + + +def _build_notification_targets(): + targets = {"log": LogChannel()} + if feishu_enabled: + targets["feishu"] = FeishuChannel.from_environment( + app_id=feishu_app_id, + app_secret=feishu_app_secret, + chat_id=feishu_chat_id, + bot_open_id=feishu_bot_open_id, + timeout_seconds=feishu_request_timeout_seconds, + ) + return targets + + +notification_targets = _build_notification_targets() +notification_channel = NotificationDispatcher(notification_targets) approval_notification_hooks = ApprovalNotificationHooks(notification_channel) -pipeline_notification_hooks = PipelineNotificationHooks(notification_channel) notification_delivery_store = NotificationDeliveryStore( merge_notification_db_path, sending_timeout_seconds=merge_notification_sending_timeout_seconds, + max_attempts=merge_notification_max_attempts, + retry_backoff_seconds=merge_notification_retry_backoff_seconds, ) -merge_notification_channel = DurableIdempotentChannel(notification_channel, notification_delivery_store) +durable_notification_targets = { + target: DurableIdempotentChannel(channel, notification_delivery_store, delivery_target=target) + for target, channel in notification_targets.items() +} +merge_notification_channel = NotificationDispatcher(durable_notification_targets) merge_notification_delivery = NotificationDelivery(merge_notification_channel, notification_delivery_store) +pipeline_notification_delivery = NotificationDelivery(merge_notification_channel, notification_delivery_store) merge_request_notification_hooks = MergeRequestNotificationHooks( merge_notification_channel, delivery=merge_notification_delivery, ) +pipeline_notification_hooks = PipelineNotificationHooks( + merge_notification_channel, + delivery=pipeline_notification_delivery, +) async def recover_merge_notification_deliveries(_app): - await merge_request_notification_hooks.recover() + recovered_merge_notifications = await merge_request_notification_hooks.recover() + recovered_pipeline_notifications = await pipeline_notification_hooks.recover() + return recovered_merge_notifications + recovered_pipeline_notifications bot.app.on_startup.append(recover_merge_notification_deliveries) diff --git a/pyproject.toml b/pyproject.toml index 798e8eb..82cb50e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ authors = [ ] requires-python = ">=3.9,<4.0" dependencies = [ + "aiohttp>=3.11,<4.0", "gidgetlab[aiohttp]>=1.1.0,<2.0.0", "setuptools<81", "langchain==0.3.6", diff --git a/src/channels/__init__.py b/src/channels/__init__.py index b919a66..f84e989 100644 --- a/src/channels/__init__.py +++ b/src/channels/__init__.py @@ -13,6 +13,14 @@ # limitations under the License. from .base import Channel +from .dispatcher import NotificationDispatcher, NotificationDispatchError +from .feishu import FeishuChannel from .log import LogChannel -__all__ = ["Channel", "LogChannel"] +__all__ = [ + "Channel", + "FeishuChannel", + "LogChannel", + "NotificationDispatchError", + "NotificationDispatcher", +] diff --git a/src/channels/dispatcher.py b/src/channels/dispatcher.py new file mode 100644 index 0000000..ac21690 --- /dev/null +++ b/src/channels/dispatcher.py @@ -0,0 +1,71 @@ +# 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 asyncio +import json +import logging +from typing import Dict, Mapping, Optional + +from ..notifications.model import Notification +from .base import Channel + + +class NotificationDispatchError(RuntimeError): + """Raised after all notification targets have had a chance to process a notification.""" + + def __init__(self, errors: Mapping[str, BaseException]): + self.errors = dict(errors) + summary = ", ".join(f"{target}: {type(error).__name__}" for target, error in self.errors.items()) + super().__init__(f"notification dispatch failed ({summary})") + + +class NotificationDispatcher(Channel): + """Send one normalized notification to independent named targets.""" + + def __init__(self, channels: Mapping[str, Channel], logger: Optional[logging.Logger] = None): + if not channels: + raise ValueError("NotificationDispatcher requires at least one target") + self.channels = dict(channels) + self.logger = logger or logging.getLogger(__name__) + + async def send(self, notification: Notification) -> None: + target_names = tuple(self.channels) + results = await asyncio.gather( + *(self.channels[target].send(notification) for target in target_names), + return_exceptions=True, + ) + errors: Dict[str, BaseException] = {} + for target, result in zip(target_names, results): + if isinstance(result, asyncio.CancelledError): + raise result + if isinstance(result, BaseException): + errors[target] = result + self.logger.error( + json.dumps( + { + "event": "notification_channel_failed", + "target": target, + "notification_action": notification.action, + "error_type": type(result).__name__, + "error": str(result), + }, + ensure_ascii=False, + sort_keys=True, + ) + ) + if errors: + raise NotificationDispatchError(errors) + + +__all__ = ["NotificationDispatchError", "NotificationDispatcher"] diff --git a/src/channels/feishu.py b/src/channels/feishu.py new file mode 100644 index 0000000..b3e5149 --- /dev/null +++ b/src/channels/feishu.py @@ -0,0 +1,319 @@ +# 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. + +# ruff: noqa: RUF001 + +import asyncio +import json +import logging +import time +from dataclasses import dataclass +from typing import Any, Callable, Dict, Mapping, Optional + +import aiohttp + +from ..notifications.model import MergeRequestNotification, Notification, PipelineNotification +from .base import Channel + +TOKEN_URL = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal" +MESSAGE_URL = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id" + + +class FeishuError(RuntimeError): + """Base error for a failed Feishu notification operation.""" + + +class FeishuConfigError(FeishuError): + """Raised when Feishu configuration is incomplete or invalid.""" + + +class FeishuTransportError(FeishuError): + """Raised when a Feishu request cannot reach the service.""" + + +class FeishuHTTPError(FeishuError): + """Raised when Feishu returns an unsuccessful HTTP status.""" + + +class FeishuResponseError(FeishuError): + """Raised when a Feishu response is malformed or reports a business error.""" + + +class FeishuAuthenticationError(FeishuResponseError): + """Raised when tenant access token acquisition fails.""" + + +@dataclass(frozen=True) +class FeishuConfig: + app_id: str + app_secret: str + chat_id: str + bot_open_id: Optional[str] = None + timeout_seconds: float = 10.0 + + +def _display(value: Any, fallback: str = "—") -> str: + if value is None or value == "": + return fallback + return str(value) + + +def _project_label(notification: Notification) -> str: + project = notification.project + return _display( + project.get("path") + or project.get("path_with_namespace") + or project.get("name") + or project.get("id") + ) + + +def _format_duration(value: Any) -> str: + if value is None or value == "": + return "—" + try: + number = float(value) + except (TypeError, ValueError): + return str(value) + if number.is_integer(): + return f"{int(number)}s" + return f"{number:g}s" + + +def _actor_label(actor: Mapping[str, Any]) -> str: + name = actor.get("name") or actor.get("username") or actor.get("id") + username = actor.get("username") + if name is None: + return "—" + if username and str(username) != str(name): + return f"{name} (@{username})" + return str(name) + + +def format_notification(notification: Notification) -> str: + """Format a normalized notification without reading a GitLab webhook payload.""" + + if isinstance(notification, MergeRequestNotification): + action_labels = { + "approval": ("✅", "已批准"), + "unapproval": ("↩️", "撤销批准"), + "merged": ("🔀", "已合并"), + } + emoji, action = action_labels.get(notification.action, ("ℹ️", notification.action or "变更")) + merge_request = notification.merge_request + return "\n".join( + ( + f"{emoji} [{_project_label(notification)}] MR {action}", + f"编号：!{_display(merge_request.get('iid'))}", + f"标题：{_display(merge_request.get('title'))}", + f"操作人：{_actor_label(notification.actor)}", + f"链接：{_display(merge_request.get('url'))}", + ) + ) + + if isinstance(notification, PipelineNotification): + pipeline = notification.pipeline + status = notification.status or pipeline.get("status") or "unknown" + emoji = {"success": "✅", "failed": "❌"}.get(status, "ℹ️") + lines = [ + f"{emoji} [{_project_label(notification)}] Pipeline {status}", + f"分支：{_display(pipeline.get('ref'))}", + f"耗时：{_format_duration(pipeline.get('duration'))}", + ] + merge_request = notification.merge_request + if merge_request: + lines.append(f"关联 MR：!{_display(merge_request.get('iid'))} {_display(merge_request.get('title'))}") + lines.append(f"链接：{_display(pipeline.get('url'))}") + return "\n".join(lines) + + raise TypeError(f"unsupported notification type: {type(notification).__name__}") + + +class FeishuChannel(Channel): + """Send normalized notifications to one Feishu chat as text messages.""" + + def __init__( + self, + app_id: str, + app_secret: str, + chat_id: str, + *, + bot_open_id: Optional[str] = None, + timeout_seconds: float = 10.0, + session_factory: Optional[Callable[..., aiohttp.ClientSession]] = None, + clock: Optional[Callable[[], float]] = None, + logger: Optional[logging.Logger] = None, + ): + missing = [ + name + for name, value in ( + ("FEISHU_APP_ID", app_id), + ("FEISHU_APP_SECRET", app_secret), + ("FEISHU_CHAT_ID", chat_id), + ) + if not value + ] + if missing: + raise FeishuConfigError(f"missing Feishu configuration: {', '.join(missing)}") + if timeout_seconds <= 0: + raise FeishuConfigError("FEISHU_REQUEST_TIMEOUT_SECONDS must be greater than zero") + + self.config = FeishuConfig( + app_id=str(app_id), + app_secret=str(app_secret), + chat_id=str(chat_id), + bot_open_id=str(bot_open_id) if bot_open_id else None, + timeout_seconds=float(timeout_seconds), + ) + self._session_factory = session_factory or aiohttp.ClientSession + self._clock = clock or time.time + self._logger = logger or logging.getLogger(__name__) + self._token: Optional[str] = None + self._token_expires_at = 0.0 + self._token_lock: Optional[asyncio.Lock] = None + self.last_message_id: Optional[str] = None + + @classmethod + def from_environment( + cls, + *, + app_id: Optional[str], + app_secret: Optional[str], + chat_id: Optional[str], + bot_open_id: Optional[str] = None, + timeout_seconds: float = 10.0, + session_factory: Optional[Callable[..., aiohttp.ClientSession]] = None, + logger: Optional[logging.Logger] = None, + ) -> "FeishuChannel": + return cls( + app_id or "", + app_secret or "", + chat_id or "", + bot_open_id=bot_open_id, + timeout_seconds=timeout_seconds, + session_factory=session_factory, + logger=logger, + ) + + async def _post_json( + self, + url: str, + *, + payload: Mapping[str, Any], + headers: Optional[Mapping[str, str]] = None, + ) -> Dict[str, Any]: + timeout = aiohttp.ClientTimeout(total=self.config.timeout_seconds) + try: + async with self._session_factory(timeout=timeout) as session: + async with session.post(url, json=dict(payload), headers=dict(headers or {})) as response: + try: + data = await response.json(content_type=None) + except (TypeError, ValueError) as exc: + raise FeishuResponseError("Feishu returned invalid JSON") from exc + if response.status < 200 or response.status >= 300: + raise FeishuHTTPError(f"Feishu HTTP request failed (status={response.status})") + if not isinstance(data, Mapping): + raise FeishuResponseError("Feishu returned an invalid response object") + return dict(data) + except FeishuError: + raise + except asyncio.TimeoutError as exc: + raise FeishuTransportError("Feishu request timed out") from exc + except aiohttp.ClientError as exc: + raise FeishuTransportError("Feishu request failed") from exc + + async def _get_tenant_access_token(self) -> str: + now = self._clock() + if self._token and self._token_expires_at > now + 60: + return self._token + + if self._token_lock is None: + self._token_lock = asyncio.Lock() + async with self._token_lock: + now = self._clock() + if self._token and self._token_expires_at > now + 60: + return self._token + + data = await self._post_json( + TOKEN_URL, + payload={"app_id": self.config.app_id, "app_secret": self.config.app_secret}, + headers={"Content-Type": "application/json"}, + ) + code = data.get("code") + token = data.get("tenant_access_token") + if code != 0 or not token: + message = _display(data.get("msg") or data.get("message"), "unknown error") + raise FeishuAuthenticationError(f"tenant access token request failed (code={code}, message={message})") + try: + expires_in = int(data.get("expire", data.get("expire_seconds", 7200))) + except (TypeError, ValueError): + expires_in = 7200 + self._token = str(token) + self._token_expires_at = self._clock() + max(expires_in, 0) + self._logger.info(json.dumps({"event": "feishu_token_refreshed"}, ensure_ascii=False)) + return self._token + + async def send(self, notification: Notification) -> None: + message = format_notification(notification) + if self.config.bot_open_id: + message = f' {message}' + + token = await self._get_tenant_access_token() + data = await self._post_json( + MESSAGE_URL, + payload={ + "receive_id": self.config.chat_id, + "msg_type": "text", + "content": json.dumps({"text": message}, ensure_ascii=False), + }, + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json; charset=utf-8", + }, + ) + code = data.get("code") + if code != 0: + error_message = _display(data.get("msg") or data.get("message"), "unknown error") + raise FeishuResponseError(f"Feishu message request failed (code={code}, message={error_message})") + + response_data = data.get("data") + message_id = response_data.get("message_id") if isinstance(response_data, Mapping) else None + self.last_message_id = str(message_id) if message_id else None + self._logger.info( + json.dumps( + { + "event": "feishu_notification_sent", + "project": _project_label(notification), + "notification_action": notification.action, + "message_id": self.last_message_id, + }, + ensure_ascii=False, + sort_keys=True, + ) + ) + + +__all__ = [ + "FeishuAuthenticationError", + "FeishuChannel", + "FeishuConfig", + "FeishuConfigError", + "FeishuError", + "FeishuHTTPError", + "FeishuResponseError", + "FeishuTransportError", + "MESSAGE_URL", + "TOKEN_URL", + "format_notification", +] diff --git a/src/config.py b/src/config.py index 08ec5f9..bc00782 100644 --- a/src/config.py +++ b/src/config.py @@ -168,3 +168,13 @@ merge_notification_sending_timeout_seconds = float( os.getenv("MERGE_NOTIFICATION_SENDING_TIMEOUT_SECONDS", "300") ) +merge_notification_max_attempts = int(os.getenv("MERGE_NOTIFICATION_MAX_ATTEMPTS", "5")) +merge_notification_retry_backoff_seconds = float(os.getenv("MERGE_NOTIFICATION_RETRY_BACKOFF_SECONDS", "1")) + +# Feishu notification channel +feishu_enabled = os.getenv("FEISHU_ENABLED", "false").lower() == "true" +feishu_app_id = os.getenv("FEISHU_APP_ID") +feishu_app_secret = os.getenv("FEISHU_APP_SECRET") +feishu_chat_id = os.getenv("FEISHU_CHAT_ID") +feishu_bot_open_id = os.getenv("FEISHU_BOT_OPEN_ID") +feishu_request_timeout_seconds = float(os.getenv("FEISHU_REQUEST_TIMEOUT_SECONDS", "10")) diff --git a/src/delivery/coordinator.py b/src/delivery/coordinator.py index b6062fc..ef8af68 100644 --- a/src/delivery/coordinator.py +++ b/src/delivery/coordinator.py @@ -18,7 +18,7 @@ from typing import Any, List, Optional from ..channels.base import Channel -from ..notifications.model import MergeRequestNotification +from ..notifications.model import Notification from .sqlite import DeliveryRecord, NotificationDeliveryStore @@ -36,21 +36,25 @@ def __init__( self.logger = logger or logging.getLogger(__name__) self.counters = Counter() - def _log(self, level: int, event: str, notification: MergeRequestNotification, **extra: Any) -> None: + def _log(self, level: int, event: str, notification: Notification, **extra: Any) -> None: + merge_request = notification.merge_request or {} + pipeline = notification.pipeline if hasattr(notification, "pipeline") else {} payload = { "event": "merge_notification_delivery", "action": event, "idempotency_key": notification.idempotency_key, "project": notification.project.get("path") or notification.project.get("id"), - "mr_iid": notification.merge_request.get("iid"), + "mr_iid": merge_request.get("iid"), + "pipeline_id": pipeline.get("id"), + "notification_type": type(notification).__name__, } payload.update(extra) self.logger.log(level, json.dumps(payload, ensure_ascii=False, sort_keys=True)) self.counters[event] += 1 - async def deliver(self, notification: MergeRequestNotification) -> bool: + async def deliver(self, notification: Notification, *, force: bool = False) -> bool: try: - decision = self.store.begin_delivery(notification) + decision = self.store.begin_delivery(notification, force=force) except Exception as exc: self.counters["failed"] += 1 self.logger.error("notification delivery state failed: %s", exc, exc_info=True) @@ -87,7 +91,7 @@ async def recover(self) -> int: async def replay_failed(self) -> int: replayed = 0 for record in self.store.failed_deliveries(): - if await self.deliver(record.notification): + if await self.deliver(record.notification, force=True): replayed += 1 if replayed: self.logger.info( diff --git a/src/delivery/idempotent_channel.py b/src/delivery/idempotent_channel.py index 8e18ee1..c0947e2 100644 --- a/src/delivery/idempotent_channel.py +++ b/src/delivery/idempotent_channel.py @@ -13,26 +13,39 @@ # limitations under the License. from ..channels.base import Channel -from ..notifications.model import MergeRequestNotification +from ..notifications.model import Notification from .sqlite import NotificationDeliveryStore class DurableIdempotentChannel(Channel): """Add a durable idempotency ledger around an existing Channel.""" - def __init__(self, channel: Channel, store: NotificationDeliveryStore): + def __init__(self, channel: Channel, store: NotificationDeliveryStore, delivery_target: str = "log"): self.channel = channel self.store = store + self.delivery_target = delivery_target - async def send(self, notification: MergeRequestNotification) -> None: - if not self.store.claim_channel_effect(notification): + async def send(self, notification: Notification) -> None: + if not self.store.claim_channel_effect(notification, delivery_target=self.delivery_target): return try: await self.channel.send(notification) except Exception as exc: - self.store.release_channel_effect(notification.idempotency_key, str(exc)) + self.store.release_channel_effect( + notification.idempotency_key, + str(exc), + delivery_target=self.delivery_target, + ) raise - self.store.mark_channel_accepted(notification.idempotency_key) + self.store.mark_channel_accepted( + notification.idempotency_key, + delivery_target=self.delivery_target, + message_id=getattr(self.channel, "last_message_id", None), + ) - def reconcile(self, notification: MergeRequestNotification) -> None: - self.store.mark_channel_accepted(notification.idempotency_key) + def reconcile(self, notification: Notification) -> None: + self.store.mark_channel_accepted( + notification.idempotency_key, + delivery_target=self.delivery_target, + message_id=getattr(self.channel, "last_message_id", None), + ) diff --git a/src/delivery/sqlite.py b/src/delivery/sqlite.py index 4d0de9a..2896d81 100644 --- a/src/delivery/sqlite.py +++ b/src/delivery/sqlite.py @@ -19,17 +19,17 @@ from pathlib import Path from typing import Any, Callable, Dict, List, Optional -from ..notifications.model import MergeRequestNotification +from ..notifications.model import MergeRequestNotification, Notification, PipelineNotification DELIVERY_STATES = frozenset(("pending", "sending", "sent", "failed")) -CHANNEL_EFFECT_STATES = frozenset(("reserved", "accepted", "failed")) +CHANNEL_EFFECT_STATES = frozenset(("reserved", "accepted", "failed", "unknown")) @dataclass(frozen=True) class DeliveryRecord: idempotency_key: str status: str - notification: MergeRequestNotification + notification: Notification attempts: int updated_at: float last_error: Optional[str] @@ -41,12 +41,31 @@ class DeliveryDecision: reason: str -def _serialize_notification(notification: MergeRequestNotification) -> str: - return json.dumps(asdict(notification), ensure_ascii=False, sort_keys=True) +def _serialize_notification(notification: Notification) -> str: + data = asdict(notification) + data["_notification_type"] = "pipeline" if isinstance(notification, PipelineNotification) else "merge_request" + return json.dumps(data, ensure_ascii=False, sort_keys=True) -def _deserialize_notification(payload: str) -> MergeRequestNotification: +def _deserialize_notification(payload: str) -> Notification: data = json.loads(payload) + notification_type = data.pop("_notification_type", None) + if notification_type == "pipeline" or (notification_type is None and "pipeline" in data): + return PipelineNotification( + source=data["source"], + event_type=data["event_type"], + action=data["action"], + webhook_action=data["webhook_action"], + status=data["status"], + message=data["message"], + project=data["project"], + pipeline=data["pipeline"], + actor=data["actor"], + occurred_at=data.get("occurred_at"), + merge_request=data.get("merge_request"), + raw_payload=data.get("raw_payload"), + idempotency_key=data.get("idempotency_key"), + ) return MergeRequestNotification( source=data["source"], event_type=data["event_type"], @@ -71,10 +90,14 @@ def __init__( path: str, *, sending_timeout_seconds: float = 300.0, + max_attempts: int = 5, + retry_backoff_seconds: float = 0.0, clock: Optional[Callable[[], float]] = None, ): self.path = path self.sending_timeout_seconds = sending_timeout_seconds + self.max_attempts = max(1, int(max_attempts)) + self.retry_backoff_seconds = max(0.0, float(retry_backoff_seconds)) self._clock = clock or time.time def _connect(self) -> sqlite3.Connection: @@ -100,19 +123,70 @@ def _connect(self) -> sqlite3.Connection: last_error TEXT ); CREATE TABLE IF NOT EXISTS channel_effects ( - idempotency_key TEXT PRIMARY KEY, - status TEXT NOT NULL CHECK(status IN ('reserved', 'accepted', 'failed')), + idempotency_key TEXT NOT NULL, + delivery_target TEXT NOT NULL DEFAULT 'log', + status TEXT NOT NULL CHECK(status IN ('reserved', 'accepted', 'failed', 'unknown')), notification_json TEXT NOT NULL, created_at REAL NOT NULL, updated_at REAL NOT NULL, - last_error TEXT + last_error TEXT, + message_id TEXT, + PRIMARY KEY (idempotency_key, delivery_target) ); CREATE INDEX IF NOT EXISTS idx_notification_deliveries_status ON notification_deliveries(status); """ ) + delivery_columns = {row[1] for row in connection.execute("PRAGMA table_info(notification_deliveries)")} + if "next_attempt_at" not in delivery_columns: + connection.execute("ALTER TABLE notification_deliveries ADD COLUMN next_attempt_at REAL") + self._migrate_legacy_channel_effects(connection) + connection.execute( + "CREATE INDEX IF NOT EXISTS idx_channel_effects_target_status " + "ON channel_effects(delivery_target, status)" + ) return connection + @staticmethod + def _migrate_legacy_channel_effects(connection: sqlite3.Connection) -> None: + columns = {row[1] for row in connection.execute("PRAGMA table_info(channel_effects)").fetchall()} + if columns and "delivery_target" not in columns: + legacy_table = "channel_effects_legacy" + legacy_exists = connection.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", + (legacy_table,), + ).fetchone() + if legacy_exists is None: + connection.execute("ALTER TABLE channel_effects RENAME TO channel_effects_legacy") + else: + connection.execute("ALTER TABLE channel_effects RENAME TO channel_effects_legacy_v1") + legacy_table = "channel_effects_legacy_v1" + connection.execute( + """ + CREATE TABLE channel_effects ( + idempotency_key TEXT NOT NULL, + delivery_target TEXT NOT NULL DEFAULT 'log', + status TEXT NOT NULL CHECK(status IN ('reserved', 'accepted', 'failed', 'unknown')), + notification_json TEXT NOT NULL, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + last_error TEXT, + message_id TEXT, + PRIMARY KEY (idempotency_key, delivery_target) + ) + """ + ) + connection.execute( + f""" + INSERT OR IGNORE INTO channel_effects + (idempotency_key, delivery_target, status, notification_json, + created_at, updated_at, last_error, message_id) + SELECT idempotency_key, 'log', status, notification_json, + created_at, updated_at, last_error, NULL + FROM {legacy_table} + """ + ) + @staticmethod def _record(row: sqlite3.Row) -> DeliveryRecord: return DeliveryRecord( @@ -124,7 +198,7 @@ def _record(row: sqlite3.Row) -> DeliveryRecord: last_error=row["last_error"], ) - def begin_delivery(self, notification: MergeRequestNotification) -> DeliveryDecision: + def begin_delivery(self, notification: Notification, *, force: bool = False) -> DeliveryDecision: key = notification.idempotency_key if not key: raise ValueError("notification idempotency_key is required") @@ -159,11 +233,20 @@ def begin_delivery(self, notification: MergeRequestNotification) -> DeliveryDeci connection.commit() return DeliveryDecision(False, "in_flight") + if row["status"] == "failed" and not force: + if row["attempts"] >= self.max_attempts: + connection.commit() + return DeliveryDecision(False, "retry_exhausted") + next_attempt_at = row["next_attempt_at"] + if next_attempt_at is not None and now < next_attempt_at: + connection.commit() + return DeliveryDecision(False, "retry_not_due") + connection.execute( """ UPDATE notification_deliveries SET status = 'sending', notification_json = ?, attempts = attempts + 1, - updated_at = ?, claimed_at = ?, last_error = NULL + updated_at = ?, claimed_at = ?, last_error = NULL, next_attempt_at = NULL WHERE idempotency_key = ? """, (_serialize_notification(notification), now, now, key), @@ -183,7 +266,8 @@ def mark_sent(self, idempotency_key: str) -> None: connection.execute( """ UPDATE notification_deliveries - SET status = 'sent', updated_at = ?, claimed_at = NULL, last_error = NULL + SET status = 'sent', updated_at = ?, claimed_at = NULL, + last_error = NULL, next_attempt_at = NULL WHERE idempotency_key = ? AND status != 'sent' """, (now, idempotency_key), @@ -195,13 +279,20 @@ def mark_failed(self, idempotency_key: str, error: str) -> None: now = self._clock() connection = self._connect() try: + row = connection.execute( + "SELECT attempts FROM notification_deliveries WHERE idempotency_key = ?", + (idempotency_key,), + ).fetchone() + attempts = int(row["attempts"]) if row is not None else 1 + delay = min(self.retry_backoff_seconds * (2 ** max(attempts - 1, 0)), 300.0) connection.execute( """ UPDATE notification_deliveries - SET status = 'failed', updated_at = ?, claimed_at = NULL, last_error = ? + SET status = 'failed', updated_at = ?, claimed_at = NULL, + last_error = ?, next_attempt_at = ? WHERE idempotency_key = ? AND status != 'sent' """, - (now, error, idempotency_key), + (now, error, now + delay, idempotency_key), ) finally: connection.close() @@ -225,10 +316,11 @@ def recoverable_deliveries(self) -> List[DeliveryRecord]: """ SELECT * FROM notification_deliveries WHERE status IN ('pending', 'failed') + AND (next_attempt_at IS NULL OR next_attempt_at <= ?) OR (status = 'sending' AND (claimed_at IS NULL OR claimed_at <= ?)) ORDER BY created_at ASC """, - (threshold,), + (self._clock(), threshold), ).fetchall() return [self._record(row) for row in rows] finally: @@ -244,26 +336,31 @@ def failed_deliveries(self) -> List[DeliveryRecord]: finally: connection.close() - def claim_channel_effect(self, notification: MergeRequestNotification) -> bool: + def claim_channel_effect(self, notification: Notification, delivery_target: str = "log") -> bool: key = notification.idempotency_key if not key: raise ValueError("notification idempotency_key is required") + if not delivery_target: + raise ValueError("delivery_target is required") now = self._clock() connection = self._connect() try: connection.execute("BEGIN IMMEDIATE") row = connection.execute( - "SELECT status FROM channel_effects WHERE idempotency_key = ?", - (key,), + """ + SELECT status FROM channel_effects + WHERE idempotency_key = ? AND delivery_target = ? + """, + (key, delivery_target), ).fetchone() if row is None: connection.execute( """ INSERT INTO channel_effects - (idempotency_key, status, notification_json, created_at, updated_at) - VALUES (?, 'reserved', ?, ?, ?) + (idempotency_key, delivery_target, status, notification_json, created_at, updated_at) + VALUES (?, ?, 'reserved', ?, ?, ?) """, - (key, _serialize_notification(notification), now, now), + (key, delivery_target, _serialize_notification(notification), now, now), ) connection.commit() return True @@ -272,9 +369,9 @@ def claim_channel_effect(self, notification: MergeRequestNotification) -> bool: """ UPDATE channel_effects SET status = 'reserved', notification_json = ?, updated_at = ?, last_error = NULL - WHERE idempotency_key = ? + WHERE idempotency_key = ? AND delivery_target = ? """, - (_serialize_notification(notification), now, key), + (_serialize_notification(notification), now, key, delivery_target), ) connection.commit() return True @@ -286,22 +383,29 @@ def claim_channel_effect(self, notification: MergeRequestNotification) -> bool: finally: connection.close() - def mark_channel_accepted(self, idempotency_key: str) -> None: + def mark_channel_accepted( + self, + idempotency_key: str, + delivery_target: str = "log", + message_id: Optional[str] = None, + ) -> None: now = self._clock() connection = self._connect() try: connection.execute( """ UPDATE channel_effects - SET status = 'accepted', updated_at = ?, last_error = NULL - WHERE idempotency_key = ? AND status IN ('reserved', 'accepted') + SET status = 'accepted', updated_at = ?, last_error = NULL, + message_id = COALESCE(?, message_id) + WHERE idempotency_key = ? AND delivery_target = ? + AND status IN ('reserved', 'accepted', 'unknown') """, - (now, idempotency_key), + (now, message_id, idempotency_key, delivery_target), ) finally: connection.close() - def release_channel_effect(self, idempotency_key: str, error: str) -> None: + def release_channel_effect(self, idempotency_key: str, error: str, delivery_target: str = "log") -> None: now = self._clock() connection = self._connect() try: @@ -309,22 +413,51 @@ def release_channel_effect(self, idempotency_key: str, error: str) -> None: """ UPDATE channel_effects SET status = 'failed', updated_at = ?, last_error = ? - WHERE idempotency_key = ? AND status = 'reserved' + WHERE idempotency_key = ? AND delivery_target = ? AND status = 'reserved' + """, + (now, error, idempotency_key, delivery_target), + ) + finally: + connection.close() + + def mark_channel_unknown(self, idempotency_key: str, error: str, delivery_target: str = "feishu") -> None: + now = self._clock() + connection = self._connect() + try: + connection.execute( + """ + UPDATE channel_effects + SET status = 'unknown', updated_at = ?, last_error = ? + WHERE idempotency_key = ? AND delivery_target = ? AND status = 'reserved' """, - (now, error, idempotency_key), + (now, error, idempotency_key, delivery_target), ) finally: connection.close() - def get_channel_effect(self, idempotency_key: str) -> Optional[Dict[str, Any]]: + def get_channel_effect(self, idempotency_key: str, delivery_target: str = "log") -> Optional[Dict[str, Any]]: connection = self._connect() try: row = connection.execute( - "SELECT * FROM channel_effects WHERE idempotency_key = ?", - (idempotency_key,), + """ + SELECT * FROM channel_effects + WHERE idempotency_key = ? AND delivery_target = ? + """, + (idempotency_key, delivery_target), ).fetchone() if row is None: return None return dict(row) finally: connection.close() + + def get_channel_effects(self, idempotency_key: str) -> List[Dict[str, Any]]: + connection = self._connect() + try: + rows = connection.execute( + "SELECT * FROM channel_effects WHERE idempotency_key = ? ORDER BY delivery_target", + (idempotency_key,), + ).fetchall() + return [dict(row) for row in rows] + finally: + connection.close() diff --git a/src/hooks/pipeline_notification.py b/src/hooks/pipeline_notification.py index 773655c..8413721 100644 --- a/src/hooks/pipeline_notification.py +++ b/src/hooks/pipeline_notification.py @@ -202,9 +202,10 @@ def build_pipeline_notification(data: Mapping[str, Any]) -> PipelineNotification class PipelineNotificationHooks: """Handle successful and failed GitLab Pipeline webhooks.""" - def __init__(self, channel: Channel, logger: Optional[logging.Logger] = None): + def __init__(self, channel: Channel, logger: Optional[logging.Logger] = None, delivery=None): self.channel = channel self.logger = logger or logging.getLogger(__name__) + self.delivery = delivery async def handle(self, event, *args, **kwargs) -> None: try: @@ -226,6 +227,10 @@ async def handle(self, event, *args, **kwargs) -> None: self.logger.error("invalid pipeline webhook: %s", exc) return + if self.delivery is not None: + await self.delivery.deliver(notification) + return + try: await self.channel.send(notification) except Exception as exc: @@ -237,3 +242,13 @@ async def handle(self, event, *args, **kwargs) -> None: exc, exc_info=True, ) + + async def recover(self) -> int: + if self.delivery is None: + return 0 + return await self.delivery.recover() + + async def replay_failed(self) -> int: + if self.delivery is None: + return 0 + return await self.delivery.replay_failed() diff --git a/tests/test_feishu_channel.py b/tests/test_feishu_channel.py new file mode 100644 index 0000000..f4641ae --- /dev/null +++ b/tests/test_feishu_channel.py @@ -0,0 +1,175 @@ +import asyncio +import json +from types import SimpleNamespace + +import pytest + +from src.channels.feishu import ( + MESSAGE_URL, + TOKEN_URL, + FeishuChannel, + FeishuConfigError, + FeishuResponseError, + format_notification, +) +from src.hooks.approval_notification import build_notification +from src.hooks.merge_notification import build_merged_notification +from src.hooks.pipeline_notification import build_pipeline_notification +from tests.fixtures.approval_webhook import copy_webhook +from tests.fixtures.merge_webhook import copy_merge_webhook +from tests.fixtures.pipeline_webhook import copy_pipeline_webhook + + +class FakeResponse: + def __init__(self, status, data): + self.status = status + self.data = data + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + async def json(self, **_kwargs): + return self.data + + +class FakeSession: + def __init__(self, responses, calls): + self.responses = responses + self.calls = calls + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + def post(self, url, **kwargs): + self.calls.append((url, kwargs)) + status, data = self.responses.pop(0) + return FakeResponse(status, data) + + +class FakeSessionFactory: + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + + def __call__(self, **_kwargs): + return FakeSession(self.responses, self.calls) + + +def make_channel(responses, **kwargs): + factory = FakeSessionFactory(responses) + return ( + FeishuChannel( + "app-id", + "app-secret", + "chat-id", + session_factory=factory, + **kwargs, + ), + factory, + ) + + +def test_feishu_channel_sends_normalized_pipeline_text_and_caches_token(): + channel, factory = make_channel( + [ + (200, {"code": 0, "tenant_access_token": "token-value", "expire": 7200}), + (200, {"code": 0, "data": {"message_id": "message-1"}}), + (200, {"code": 0, "data": {"message_id": "message-2"}}), + ] + ) + notification = build_pipeline_notification(copy_pipeline_webhook(status="failed")) + + async def run(): + await channel.send(notification) + await channel.send(notification) + + asyncio.run(run()) + + assert [call[0] for call in factory.calls] == [TOKEN_URL, MESSAGE_URL, MESSAGE_URL] + message_payload = factory.calls[1][1]["json"] + assert message_payload["receive_id"] == "chat-id" + assert message_payload["msg_type"] == "text" + message_text = json.loads(message_payload["content"])["text"] + assert "❌" in message_text + assert "Pipeline failed" in message_text + assert "master" in message_text + assert channel.last_message_id == "message-2" + + +def test_feishu_channel_refreshes_token_when_expiring(): + now = [1000.0] + channel, factory = make_channel( + [ + (200, {"code": 0, "tenant_access_token": "first-token", "expire": 61}), + (200, {"code": 0, "data": {}}), + (200, {"code": 0, "tenant_access_token": "second-token", "expire": 7200}), + (200, {"code": 0, "data": {}}), + ], + clock=lambda: now[0], + ) + notification = build_pipeline_notification(copy_pipeline_webhook(status="success")) + + async def run(): + await channel.send(notification) + now[0] = 1001.0 + await channel.send(notification) + + asyncio.run(run()) + + assert [call[0] for call in factory.calls] == [TOKEN_URL, MESSAGE_URL, TOKEN_URL, MESSAGE_URL] + assert factory.calls[3][1]["headers"]["Authorization"] == "Bearer second-token" + + +def test_feishu_channel_supports_optional_at_and_mr_formatting(): + channel, factory = make_channel( + [ + (200, {"code": 0, "tenant_access_token": "token-value", "expire": 7200}), + (200, {"code": 0, "data": {}}), + ], + bot_open_id="ou_test-user", + ) + notification = build_merged_notification(copy_merge_webhook()) + + asyncio.run(channel.send(notification)) + + message_text = json.loads(factory.calls[1][1]["json"]["content"])["text"] + assert '' in message_text + assert "MR" in message_text + assert "!12" in message_text + + +def test_feishu_channel_rejects_business_error_without_logging_secret(): + channel, _factory = make_channel( + [ + (200, {"code": 999, "msg": "permission denied"}), + ] + ) + + with pytest.raises(FeishuResponseError, match="code=999"): + asyncio.run(channel.send(build_pipeline_notification(copy_pipeline_webhook(status="failed")))) + + +def test_feishu_channel_requires_complete_configuration(): + with pytest.raises(FeishuConfigError, match="FEISHU_APP_SECRET"): + FeishuChannel("app-id", "", "chat-id") + + +def test_format_notification_rejects_unknown_normalized_type(): + with pytest.raises(TypeError, match="unsupported notification type"): + format_notification(SimpleNamespace(action="unknown")) + + +@pytest.mark.parametrize( + ("webhook_action", "expected"), + (("approval", "已批准"), ("unapproval", "撤销批准")), +) +def test_format_notification_supports_approval_and_unapproval(webhook_action, expected): + notification = build_notification(copy_webhook(action=webhook_action)) + + assert expected in format_notification(notification) diff --git a/tests/test_feishu_delivery.py b/tests/test_feishu_delivery.py new file mode 100644 index 0000000..051d44d --- /dev/null +++ b/tests/test_feishu_delivery.py @@ -0,0 +1,141 @@ +import asyncio +import json +import sqlite3 +from dataclasses import asdict +from types import SimpleNamespace + +from src.channels.dispatcher import NotificationDispatcher +from src.delivery.coordinator import NotificationDelivery +from src.delivery.idempotent_channel import DurableIdempotentChannel +from src.delivery.sqlite import NotificationDeliveryStore +from src.hooks.merge_notification import build_merged_notification +from src.hooks.pipeline_notification import PipelineNotificationHooks, build_pipeline_notification +from src.notifications.model import PipelineNotification +from tests.fixtures.merge_webhook import copy_merge_webhook +from tests.fixtures.pipeline_webhook import copy_pipeline_webhook + + +def test_feishu_is_not_constructed_when_disabled(monkeypatch): + import gitlab_bot + + monkeypatch.setattr(gitlab_bot, "feishu_enabled", False) + + targets = gitlab_bot._build_notification_targets() + + assert tuple(targets) == ("log",) + + +class RecordingChannel: + def __init__(self): + self.calls = 0 + self.notifications = [] + + async def send(self, notification): + self.calls += 1 + self.notifications.append(notification) + + +class FailingOnceChannel(RecordingChannel): + def __init__(self): + super().__init__() + self.failed = False + + async def send(self, notification): + self.calls += 1 + if not self.failed: + self.failed = True + raise RuntimeError("temporary Feishu failure") + self.notifications.append(notification) + + +def make_durable_dispatcher(store, log_channel, feishu_channel): + return NotificationDispatcher( + { + "log": DurableIdempotentChannel(log_channel, store, delivery_target="log"), + "feishu": DurableIdempotentChannel(feishu_channel, store, delivery_target="feishu"), + } + ) + + +def test_pipeline_delivery_keeps_log_success_when_feishu_retries(tmp_path): + log_channel = RecordingChannel() + feishu_channel = FailingOnceChannel() + store = NotificationDeliveryStore(str(tmp_path / "delivery.sqlite3")) + dispatcher = make_durable_dispatcher(store, log_channel, feishu_channel) + delivery = NotificationDelivery(dispatcher, store) + notification = build_pipeline_notification(copy_pipeline_webhook(status="failed")) + + first = asyncio.run(delivery.deliver(notification)) + first_record = store.get_delivery(notification.idempotency_key) + first_log_effect = store.get_channel_effect(notification.idempotency_key, "log") + first_feishu_effect = store.get_channel_effect(notification.idempotency_key, "feishu") + second = asyncio.run(delivery.deliver(notification)) + + assert first is False + assert second is True + assert first_record.status == "failed" + assert first_log_effect["status"] == "accepted" + assert first_feishu_effect["status"] == "failed" + assert log_channel.calls == 1 + assert feishu_channel.calls == 2 + assert store.get_channel_effect(notification.idempotency_key, "feishu")["status"] == "accepted" + assert store.get_delivery(notification.idempotency_key).status == "sent" + + +def test_pipeline_hook_uses_durable_delivery_and_serializes_pipeline_notification(tmp_path): + log_channel = RecordingChannel() + feishu_channel = RecordingChannel() + store = NotificationDeliveryStore(str(tmp_path / "delivery.sqlite3")) + dispatcher = make_durable_dispatcher(store, log_channel, feishu_channel) + delivery = NotificationDelivery(dispatcher, store) + hooks = PipelineNotificationHooks(dispatcher, delivery=delivery) + event = SimpleNamespace(data=copy_pipeline_webhook(status="success")) + + asyncio.run(hooks.handle(event)) + + notification = log_channel.notifications[0] + record = store.get_delivery(notification.idempotency_key) + assert isinstance(record.notification, PipelineNotification) + assert record.notification.status == "success" + assert log_channel.calls == 1 + assert feishu_channel.calls == 1 + + +def test_pipeline_hook_without_delivery_keeps_direct_channel_contract(): + channel = RecordingChannel() + hooks = PipelineNotificationHooks(channel) + + asyncio.run(hooks.handle(SimpleNamespace(data=copy_pipeline_webhook(status="success")))) + + assert channel.calls == 1 + + +def test_legacy_single_target_effects_are_migrated_as_log(tmp_path): + notification = build_merged_notification(copy_merge_webhook()) + database_path = tmp_path / "legacy.sqlite3" + connection = sqlite3.connect(database_path) + connection.executescript( + """ + CREATE TABLE channel_effects ( + idempotency_key TEXT PRIMARY KEY, + status TEXT NOT NULL CHECK(status IN ('reserved', 'accepted', 'failed')), + notification_json TEXT NOT NULL, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + last_error TEXT + ); + """ + ) + connection.execute( + "INSERT INTO channel_effects VALUES (?, 'accepted', ?, 1, 1, NULL)", + (notification.idempotency_key, json.dumps(asdict(notification))), + ) + connection.commit() + connection.close() + + store = NotificationDeliveryStore(str(database_path)) + effect = store.get_channel_effect(notification.idempotency_key, "log") + + assert effect["delivery_target"] == "log" + assert effect["status"] == "accepted" + assert store.get_channel_effect(notification.idempotency_key, "feishu") is None diff --git a/tests/test_notification_delivery.py b/tests/test_notification_delivery.py index ed35426..470d86c 100644 --- a/tests/test_notification_delivery.py +++ b/tests/test_notification_delivery.py @@ -117,6 +117,39 @@ async def run(): assert store.get_delivery(notification.idempotency_key).attempts == 2 +def test_retry_backoff_defers_automatic_retry_until_due(tmp_path): + channel = FailingOnceChannel() + now = [1000.0] + store = NotificationDeliveryStore( + str(tmp_path / "delivery.sqlite3"), + retry_backoff_seconds=10, + clock=lambda: now[0], + ) + delivery = NotificationDelivery(DurableIdempotentChannel(channel, store), store) + notification = make_notification() + + assert asyncio.run(delivery.deliver(notification)) is False + assert asyncio.run(delivery.deliver(notification)) is False + assert channel.calls == 1 + + now[0] += 10 + assert asyncio.run(delivery.deliver(notification)) is True + assert channel.calls == 2 + + +def test_retry_attempts_are_bounded_but_manual_replay_can_continue(tmp_path): + channel = FailingOnceChannel() + store = NotificationDeliveryStore(str(tmp_path / "delivery.sqlite3"), max_attempts=1) + delivery = NotificationDelivery(DurableIdempotentChannel(channel, store), store) + notification = make_notification() + + assert asyncio.run(delivery.deliver(notification)) is False + assert asyncio.run(delivery.deliver(notification)) is False + assert channel.calls == 1 + assert asyncio.run(delivery.replay_failed()) == 1 + assert channel.calls == 2 + + def test_restart_recovers_failed_delivery_and_manual_replay(tmp_path): first_channel = FailingOnceChannel() store, first_delivery = make_delivery(tmp_path, first_channel) diff --git a/tests/test_notification_dispatcher.py b/tests/test_notification_dispatcher.py new file mode 100644 index 0000000..61a2506 --- /dev/null +++ b/tests/test_notification_dispatcher.py @@ -0,0 +1,42 @@ +import asyncio + +import pytest + +from src.channels.dispatcher import NotificationDispatcher, NotificationDispatchError +from src.hooks.pipeline_notification import build_pipeline_notification +from tests.fixtures.pipeline_webhook import copy_pipeline_webhook + + +class RecordingChannel: + def __init__(self, error=None): + self.notifications = [] + self.error = error + + async def send(self, notification): + self.notifications.append(notification) + if self.error: + raise self.error + + +def test_dispatcher_runs_all_targets_and_preserves_partial_success(): + log_channel = RecordingChannel() + feishu_channel = RecordingChannel(RuntimeError("feishu unavailable")) + dispatcher = NotificationDispatcher({"log": log_channel, "feishu": feishu_channel}) + notification = build_pipeline_notification(copy_pipeline_webhook(status="failed")) + + with pytest.raises(NotificationDispatchError) as error: + asyncio.run(dispatcher.send(notification)) + + assert log_channel.notifications == [notification] + assert feishu_channel.notifications == [notification] + assert set(error.value.errors) == {"feishu"} + + +def test_dispatcher_with_one_target_keeps_log_only_mode(): + log_channel = RecordingChannel() + dispatcher = NotificationDispatcher({"log": log_channel}) + notification = build_pipeline_notification(copy_pipeline_webhook(status="success")) + + asyncio.run(dispatcher.send(notification)) + + assert log_channel.notifications == [notification] diff --git a/uv.lock b/uv.lock index 6f9cda6..ae90c8e 100644 --- a/uv.lock +++ b/uv.lock @@ -446,6 +446,7 @@ name = "gitlab-bot" version = "1.2.1" source = { editable = "." } dependencies = [ + { name = "aiohttp" }, { name = "gidgetlab", extra = ["aiohttp"] }, { name = "langchain" }, { name = "langchain-google-genai" }, @@ -465,6 +466,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "aiohttp", specifier = ">=3.11,<4.0" }, { name = "gidgetlab", extras = ["aiohttp"], specifier = ">=1.1.0,<2.0.0" }, { name = "langchain", specifier = "==0.3.6" }, { name = "langchain-google-genai", specifier = "==2.0.4" },