From a97d0d369dc263b52516fec69d21d792badeadf1 Mon Sep 17 00:00:00 2001 From: Lei Zhang Date: Sun, 2 Aug 2026 20:09:17 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E6=88=90=20MR=20=E9=80=9A=E7=9F=A5?= =?UTF-8?q?=E4=B8=8E=20Hook/Channel=20=E6=A8=A1=E5=9D=97=E9=87=8D=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gitlab_bot.py | 40 ++- src/channels/__init__.py | 18 + src/channels/base.py | 25 ++ src/channels/log.py | 31 ++ src/config.py | 9 + src/delivery/__init__.py | 13 + src/delivery/coordinator.py | 99 ++++++ src/delivery/idempotent_channel.py | 38 ++ src/delivery/sqlite.py | 330 ++++++++++++++++++ src/hooks/__init__.py | 13 + src/{ => hooks}/approval_notification.py | 49 +-- src/{issue_hook.py => hooks/issue.py} | 0 src/hooks/merge_notification.py | 274 +++++++++++++++ .../merge_request.py} | 0 src/{note_hook.py => hooks/note.py} | 0 src/locales/en/LC_MESSAGES/gitlab-bot.po | 6 +- src/locales/zh/LC_MESSAGES/gitlab-bot.po | 6 +- src/notifications/__init__.py | 17 + src/notifications/model.py | 34 ++ tests/fixtures/merge_webhook.py | 49 +++ tests/test_approval_notification.py | 7 +- tests/test_delivery_modules.py | 9 + tests/test_hook_modules.py | 9 + tests/test_merge_notification.py | 134 +++++++ tests/test_merge_request_approval_contract.py | 2 +- tests/test_merge_request_check_commit.py | 2 +- tests/test_merge_request_failure_state.py | 2 +- .../test_merge_request_webhook_regression.py | 2 +- tests/test_notification_contracts.py | 18 + tests/test_notification_delivery.py | 208 +++++++++++ tests/test_notification_module_boundaries.py | 104 ++++++ tests/test_runtime_startup.py | 6 + tests/test_webhook_hook_modules.py | 38 ++ 33 files changed, 1535 insertions(+), 57 deletions(-) create mode 100644 src/channels/__init__.py create mode 100644 src/channels/base.py create mode 100644 src/channels/log.py create mode 100644 src/delivery/__init__.py create mode 100644 src/delivery/coordinator.py create mode 100644 src/delivery/idempotent_channel.py create mode 100644 src/delivery/sqlite.py create mode 100644 src/hooks/__init__.py rename src/{ => hooks}/approval_notification.py (83%) rename src/{issue_hook.py => hooks/issue.py} (100%) create mode 100644 src/hooks/merge_notification.py rename src/{merge_request_hook.py => hooks/merge_request.py} (100%) rename src/{note_hook.py => hooks/note.py} (100%) create mode 100644 src/notifications/__init__.py create mode 100644 src/notifications/model.py create mode 100644 tests/fixtures/merge_webhook.py create mode 100644 tests/test_delivery_modules.py create mode 100644 tests/test_hook_modules.py create mode 100644 tests/test_merge_notification.py create mode 100644 tests/test_notification_contracts.py create mode 100644 tests/test_notification_delivery.py create mode 100644 tests/test_notification_module_boundaries.py create mode 100644 tests/test_webhook_hook_modules.py diff --git a/gitlab_bot.py b/gitlab_bot.py index c740641..e195f14 100644 --- a/gitlab_bot.py +++ b/gitlab_bot.py @@ -17,18 +17,25 @@ from dotenv import load_dotenv -from src.approval_notification import ApprovalNotificationHooks, LogChannel +from src.channels.log import LogChannel from src.config import ( bot_gitlab_token, bot_gitlab_url, bot_gitlab_username, bot_host, bot_port, + merge_notification_db_path, + merge_notification_sending_timeout_seconds, ) -from src.issue_hook import IssueHooks +from src.delivery.coordinator import NotificationDelivery +from src.delivery.idempotent_channel import DurableIdempotentChannel +from src.delivery.sqlite import NotificationDeliveryStore +from src.hooks.approval_notification import ApprovalNotificationHooks +from src.hooks.issue import IssueHooks +from src.hooks.merge_notification import MergeRequestNotificationHooks +from src.hooks.merge_request import MergeRequestHooks +from src.hooks.note import NoteHooks from src.logs import print_event -from src.merge_request_hook import MergeRequestHooks -from src.note_hook import NoteHooks load_dotenv() # isort:skip @@ -58,7 +65,25 @@ def _load_gitlab_bot(): issue_hooks = IssueHooks() merge_request_hooks = MergeRequestHooks() note_hooks = NoteHooks() -approval_notification_hooks = ApprovalNotificationHooks(LogChannel()) +notification_channel = LogChannel() +approval_notification_hooks = ApprovalNotificationHooks(notification_channel) +notification_delivery_store = NotificationDeliveryStore( + merge_notification_db_path, + sending_timeout_seconds=merge_notification_sending_timeout_seconds, +) +merge_notification_channel = DurableIdempotentChannel(notification_channel, notification_delivery_store) +merge_notification_delivery = NotificationDelivery(merge_notification_channel, notification_delivery_store) +merge_request_notification_hooks = MergeRequestNotificationHooks( + merge_notification_channel, + delivery=merge_notification_delivery, +) + + +async def recover_merge_notification_deliveries(_app): + await merge_request_notification_hooks.recover() + + +bot.app.on_startup.append(recover_merge_notification_deliveries) @bot.router.register("Issue Hook", action="open") @@ -115,6 +140,11 @@ async def merge_request_unapproval_event(event, gl, *args, **kwargs): await approval_notification_hooks.handle(event, gl, *args, **kwargs) +@bot.router.register("Merge Request Hook", action="merge") +async def merge_request_merged_event(event, gl, *args, **kwargs): + await merge_request_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/__init__.py b/src/channels/__init__.py new file mode 100644 index 0000000..b919a66 --- /dev/null +++ b/src/channels/__init__.py @@ -0,0 +1,18 @@ +# 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. + +from .base import Channel +from .log import LogChannel + +__all__ = ["Channel", "LogChannel"] diff --git a/src/channels/base.py b/src/channels/base.py new file mode 100644 index 0000000..574d082 --- /dev/null +++ b/src/channels/base.py @@ -0,0 +1,25 @@ +# 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. + +from abc import ABC, abstractmethod + +from ..notifications.model import MergeRequestNotification + + +class Channel(ABC): + """Asynchronous destination for normalized notifications.""" + + @abstractmethod + async def send(self, notification: MergeRequestNotification) -> None: + raise NotImplementedError diff --git a/src/channels/log.py b/src/channels/log.py new file mode 100644 index 0000000..bec1d7d --- /dev/null +++ b/src/channels/log.py @@ -0,0 +1,31 @@ +# 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 dataclasses import asdict +from typing import Optional + +from ..notifications.model import MergeRequestNotification +from .base import Channel + + +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)) diff --git a/src/config.py b/src/config.py index ba83db0..08ec5f9 100644 --- a/src/config.py +++ b/src/config.py @@ -159,3 +159,12 @@ ) bot_gitlab_merge_request_reviewer_username = os.getenv("BOT_GITLAB_MERGE_REQUEST_REVIEWER_USERNAME", None) + +# merge notification delivery +merge_notification_db_path = os.getenv( + "MERGE_NOTIFICATION_DB_PATH", + "data/merge-notifications.sqlite3", +) +merge_notification_sending_timeout_seconds = float( + os.getenv("MERGE_NOTIFICATION_SENDING_TIMEOUT_SECONDS", "300") +) diff --git a/src/delivery/__init__.py b/src/delivery/__init__.py new file mode 100644 index 0000000..9036662 --- /dev/null +++ b/src/delivery/__init__.py @@ -0,0 +1,13 @@ +# 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. diff --git a/src/delivery/coordinator.py b/src/delivery/coordinator.py new file mode 100644 index 0000000..b6062fc --- /dev/null +++ b/src/delivery/coordinator.py @@ -0,0 +1,99 @@ +# 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 collections import Counter +from typing import Any, List, Optional + +from ..channels.base import Channel +from ..notifications.model import MergeRequestNotification +from .sqlite import DeliveryRecord, NotificationDeliveryStore + + +class NotificationDelivery: + """Coordinate durable delivery state, recovery, and observability.""" + + def __init__( + self, + channel: Channel, + store: NotificationDeliveryStore, + logger: Optional[logging.Logger] = None, + ): + self.channel = channel + self.store = store + self.logger = logger or logging.getLogger(__name__) + self.counters = Counter() + + def _log(self, level: int, event: str, notification: MergeRequestNotification, **extra: Any) -> None: + 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"), + } + 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: + try: + decision = self.store.begin_delivery(notification) + except Exception as exc: + self.counters["failed"] += 1 + self.logger.error("notification delivery state failed: %s", exc, exc_info=True) + return False + if not decision.should_send: + self._log(logging.INFO, "duplicate", notification, reason=decision.reason) + return False + + try: + await self.channel.send(notification) + except Exception as exc: + self.store.mark_failed(notification.idempotency_key, str(exc)) + self._log(logging.ERROR, "failed", notification, error=str(exc)) + return False + + reconcile = getattr(self.channel, "reconcile", None) + if callable(reconcile): + reconcile(notification) + self.store.mark_sent(notification.idempotency_key) + self._log(logging.INFO, "sent", notification, reason=decision.reason) + return True + + async def recover(self) -> int: + recovered = 0 + for record in self.store.recoverable_deliveries(): + if await self.deliver(record.notification): + recovered += 1 + if recovered: + self.logger.info( + json.dumps({"event": "merge_notification_recovery", "count": recovered}, sort_keys=True) + ) + return recovered + + async def replay_failed(self) -> int: + replayed = 0 + for record in self.store.failed_deliveries(): + if await self.deliver(record.notification): + replayed += 1 + if replayed: + self.logger.info( + json.dumps({"event": "merge_notification_replay", "count": replayed}, sort_keys=True) + ) + return replayed + + def failure_backlog(self) -> List[DeliveryRecord]: + return self.store.failed_deliveries() diff --git a/src/delivery/idempotent_channel.py b/src/delivery/idempotent_channel.py new file mode 100644 index 0000000..8e18ee1 --- /dev/null +++ b/src/delivery/idempotent_channel.py @@ -0,0 +1,38 @@ +# 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. + +from ..channels.base import Channel +from ..notifications.model import MergeRequestNotification +from .sqlite import NotificationDeliveryStore + + +class DurableIdempotentChannel(Channel): + """Add a durable idempotency ledger around an existing Channel.""" + + def __init__(self, channel: Channel, store: NotificationDeliveryStore): + self.channel = channel + self.store = store + + async def send(self, notification: MergeRequestNotification) -> None: + if not self.store.claim_channel_effect(notification): + return + try: + await self.channel.send(notification) + except Exception as exc: + self.store.release_channel_effect(notification.idempotency_key, str(exc)) + raise + self.store.mark_channel_accepted(notification.idempotency_key) + + def reconcile(self, notification: MergeRequestNotification) -> None: + self.store.mark_channel_accepted(notification.idempotency_key) diff --git a/src/delivery/sqlite.py b/src/delivery/sqlite.py new file mode 100644 index 0000000..4d0de9a --- /dev/null +++ b/src/delivery/sqlite.py @@ -0,0 +1,330 @@ +# 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 sqlite3 +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional + +from ..notifications.model import MergeRequestNotification + +DELIVERY_STATES = frozenset(("pending", "sending", "sent", "failed")) +CHANNEL_EFFECT_STATES = frozenset(("reserved", "accepted", "failed")) + + +@dataclass(frozen=True) +class DeliveryRecord: + idempotency_key: str + status: str + notification: MergeRequestNotification + attempts: int + updated_at: float + last_error: Optional[str] + + +@dataclass(frozen=True) +class DeliveryDecision: + should_send: bool + reason: str + + +def _serialize_notification(notification: MergeRequestNotification) -> str: + return json.dumps(asdict(notification), ensure_ascii=False, sort_keys=True) + + +def _deserialize_notification(payload: str) -> MergeRequestNotification: + data = json.loads(payload) + return MergeRequestNotification( + source=data["source"], + event_type=data["event_type"], + action=data["action"], + webhook_action=data["webhook_action"], + message=data["message"], + project=data["project"], + merge_request=data["merge_request"], + actor=data["actor"], + occurred_at=data.get("occurred_at"), + raw_payload=data.get("raw_payload"), + triggered_by=data.get("triggered_by"), + idempotency_key=data.get("idempotency_key"), + ) + + +class NotificationDeliveryStore: + """SQLite-backed state and Channel-effect ledger for notifications.""" + + def __init__( + self, + path: str, + *, + sending_timeout_seconds: float = 300.0, + clock: Optional[Callable[[], float]] = None, + ): + self.path = path + self.sending_timeout_seconds = sending_timeout_seconds + self._clock = clock or time.time + + def _connect(self) -> sqlite3.Connection: + if self.path != ":memory:": + parent = Path(self.path).expanduser().parent + if str(parent) not in ("", "."): + parent.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect(self.path, timeout=30, isolation_level=None) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA busy_timeout=30000") + if self.path != ":memory:": + connection.execute("PRAGMA journal_mode=WAL") + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS notification_deliveries ( + idempotency_key TEXT PRIMARY KEY, + status TEXT NOT NULL CHECK(status IN ('pending', 'sending', 'sent', 'failed')), + notification_json TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + claimed_at REAL, + 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')), + notification_json TEXT NOT NULL, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + last_error TEXT + ); + CREATE INDEX IF NOT EXISTS idx_notification_deliveries_status + ON notification_deliveries(status); + """ + ) + return connection + + @staticmethod + def _record(row: sqlite3.Row) -> DeliveryRecord: + return DeliveryRecord( + idempotency_key=row["idempotency_key"], + status=row["status"], + notification=_deserialize_notification(row["notification_json"]), + attempts=row["attempts"], + updated_at=row["updated_at"], + last_error=row["last_error"], + ) + + def begin_delivery(self, notification: MergeRequestNotification) -> DeliveryDecision: + key = notification.idempotency_key + if not key: + raise ValueError("notification idempotency_key is required") + now = self._clock() + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + "SELECT * FROM notification_deliveries WHERE idempotency_key = ?", + (key,), + ).fetchone() + if row is None: + connection.execute( + """ + INSERT INTO notification_deliveries + (idempotency_key, status, notification_json, attempts, created_at, updated_at, claimed_at) + VALUES (?, 'sending', ?, 1, ?, ?, ?) + """, + (key, _serialize_notification(notification), now, now, now), + ) + connection.commit() + return DeliveryDecision(True, "new") + + if row["status"] == "sent": + connection.commit() + return DeliveryDecision(False, "already_sent") + + if row["status"] == "sending": + claimed_at = row["claimed_at"] + is_active = claimed_at is not None and now - claimed_at < self.sending_timeout_seconds + if is_active: + connection.commit() + return DeliveryDecision(False, "in_flight") + + connection.execute( + """ + UPDATE notification_deliveries + SET status = 'sending', notification_json = ?, attempts = attempts + 1, + updated_at = ?, claimed_at = ?, last_error = NULL + WHERE idempotency_key = ? + """, + (_serialize_notification(notification), now, now, key), + ) + connection.commit() + return DeliveryDecision(True, "retry") + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def mark_sent(self, idempotency_key: str) -> None: + now = self._clock() + connection = self._connect() + try: + connection.execute( + """ + UPDATE notification_deliveries + SET status = 'sent', updated_at = ?, claimed_at = NULL, last_error = NULL + WHERE idempotency_key = ? AND status != 'sent' + """, + (now, idempotency_key), + ) + finally: + connection.close() + + def mark_failed(self, idempotency_key: str, error: str) -> None: + now = self._clock() + connection = self._connect() + try: + connection.execute( + """ + UPDATE notification_deliveries + SET status = 'failed', updated_at = ?, claimed_at = NULL, last_error = ? + WHERE idempotency_key = ? AND status != 'sent' + """, + (now, error, idempotency_key), + ) + finally: + connection.close() + + def get_delivery(self, idempotency_key: str) -> Optional[DeliveryRecord]: + connection = self._connect() + try: + row = connection.execute( + "SELECT * FROM notification_deliveries WHERE idempotency_key = ?", + (idempotency_key,), + ).fetchone() + return self._record(row) if row is not None else None + finally: + connection.close() + + def recoverable_deliveries(self) -> List[DeliveryRecord]: + threshold = self._clock() - self.sending_timeout_seconds + connection = self._connect() + try: + rows = connection.execute( + """ + SELECT * FROM notification_deliveries + WHERE status IN ('pending', 'failed') + OR (status = 'sending' AND (claimed_at IS NULL OR claimed_at <= ?)) + ORDER BY created_at ASC + """, + (threshold,), + ).fetchall() + return [self._record(row) for row in rows] + finally: + connection.close() + + def failed_deliveries(self) -> List[DeliveryRecord]: + connection = self._connect() + try: + rows = connection.execute( + "SELECT * FROM notification_deliveries WHERE status = 'failed' ORDER BY updated_at ASC" + ).fetchall() + return [self._record(row) for row in rows] + finally: + connection.close() + + def claim_channel_effect(self, notification: MergeRequestNotification) -> bool: + key = notification.idempotency_key + if not key: + raise ValueError("notification idempotency_key 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,), + ).fetchone() + if row is None: + connection.execute( + """ + INSERT INTO channel_effects + (idempotency_key, status, notification_json, created_at, updated_at) + VALUES (?, 'reserved', ?, ?, ?) + """, + (key, _serialize_notification(notification), now, now), + ) + connection.commit() + return True + if row["status"] == "failed": + connection.execute( + """ + UPDATE channel_effects + SET status = 'reserved', notification_json = ?, updated_at = ?, last_error = NULL + WHERE idempotency_key = ? + """, + (_serialize_notification(notification), now, key), + ) + connection.commit() + return True + connection.commit() + return False + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def mark_channel_accepted(self, idempotency_key: str) -> 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') + """, + (now, idempotency_key), + ) + finally: + connection.close() + + def release_channel_effect(self, idempotency_key: str, error: str) -> None: + now = self._clock() + connection = self._connect() + try: + connection.execute( + """ + UPDATE channel_effects + SET status = 'failed', updated_at = ?, last_error = ? + WHERE idempotency_key = ? AND status = 'reserved' + """, + (now, error, idempotency_key), + ) + finally: + connection.close() + + def get_channel_effect(self, idempotency_key: str) -> Optional[Dict[str, Any]]: + connection = self._connect() + try: + row = connection.execute( + "SELECT * FROM channel_effects WHERE idempotency_key = ?", + (idempotency_key,), + ).fetchone() + if row is None: + return None + return dict(row) + finally: + connection.close() diff --git a/src/hooks/__init__.py b/src/hooks/__init__.py new file mode 100644 index 0000000..9036662 --- /dev/null +++ b/src/hooks/__init__.py @@ -0,0 +1,13 @@ +# 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. diff --git a/src/approval_notification.py b/src/hooks/approval_notification.py similarity index 83% rename from src/approval_notification.py rename to src/hooks/approval_notification.py index c4fe251..bbf27ac 100644 --- a/src/approval_notification.py +++ b/src/hooks/approval_notification.py @@ -12,11 +12,11 @@ # 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 +from typing import Any, Mapping, Optional + +from ..channels.base import Channel +from ..notifications.model import MergeRequestNotification APPROVAL_ACTIONS = frozenset(("approval", "approved", "unapproval", "unapproved")) NORMALIZED_ACTIONS = { @@ -26,39 +26,14 @@ "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)) +__all__ = [ + "APPROVAL_ACTIONS", + "NORMALIZED_ACTIONS", + "MergeRequestNotification", + "Channel", + "build_notification", + "ApprovalNotificationHooks", +] def _first_value(data: Mapping[str, Any], *keys: str) -> Any: diff --git a/src/issue_hook.py b/src/hooks/issue.py similarity index 100% rename from src/issue_hook.py rename to src/hooks/issue.py diff --git a/src/hooks/merge_notification.py b/src/hooks/merge_notification.py new file mode 100644 index 0000000..4482815 --- /dev/null +++ b/src/hooks/merge_notification.py @@ -0,0 +1,274 @@ +# 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 MergeRequestNotification + +MERGE_WEBHOOK_ACTION = "merge" +MERGED_NOTIFICATION_ACTION = "merged" +MERGE_NOTIFICATION_EVENT_TYPE = "merge_request_lifecycle" +AUTO_MERGE_ACTOR = { + "id": None, + "username": "gitlab-auto-merge", + "name": "GitLab 自动合并", + "is_system": True, +} + +__all__ = [ + "MERGE_WEBHOOK_ACTION", + "MERGED_NOTIFICATION_ACTION", + "MERGE_NOTIFICATION_EVENT_TYPE", + "AUTO_MERGE_ACTOR", + "MergeRequestNotification", + "Channel", + "build_merged_notification", + "MergeRequestNotificationHooks", +] + + +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 _mapping_value(data: Mapping[str, Any], *keys: str) -> Optional[Mapping[str, Any]]: + for key in keys: + value = data.get(key) + if isinstance(value, Mapping): + return value + return None + + +def _normalize_actor(data: Mapping[str, Any]) -> Dict[str, Any]: + username = _first_value(data, "username", "user_name") + name = _first_value(data, "name", "user_name") + if name is None: + name = username + return { + "id": data.get("id"), + "username": username, + "name": name, + } + + +def _merge_actor(payload: Mapping[str, Any], attributes: Mapping[str, Any]) -> Dict[str, Any]: + merge_user = _mapping_value(attributes, "merge_user", "merged_by") + if merge_user is None: + merge_user = _mapping_value(payload, "merge_user", "merged_by") + if merge_user is not None: + actor = _normalize_actor(merge_user) + if any(actor.get(key) is not None and actor.get(key) != "" for key in ("id", "username", "name")): + return actor + + merge_user_id = _first_value(attributes, "merge_user_id", "merged_by_id") + if merge_user_id is None: + merge_user_id = _first_value(payload, "merge_user_id", "merged_by_id") + merge_user_name = _first_value(attributes, "merge_user_name", "merged_by_name") + if merge_user_name is None: + merge_user_name = _first_value(payload, "merge_user_name", "merged_by_name") + merge_user_username = _first_value(attributes, "merge_user_username", "merged_by_username") + if merge_user_username is None: + merge_user_username = _first_value(payload, "merge_user_username", "merged_by_username") + if any(value is not None for value in (merge_user_id, merge_user_name, merge_user_username)): + return { + "id": merge_user_id, + "username": merge_user_username, + "name": merge_user_name or merge_user_username or str(merge_user_id), + } + + return dict(AUTO_MERGE_ACTOR) + + +def _actor_display_name(actor: Mapping[str, Any]) -> str: + name = actor.get("name") or actor.get("username") or actor.get("id") + if name is None or name == "": + return str(AUTO_MERGE_ACTOR["name"]) + username = actor.get("username") + if username and str(username) != str(name) and not actor.get("is_system"): + return f"{name} (@{username})" + return str(name) + + +def _merge_idempotency_key( + payload: Mapping[str, Any], + attributes: Mapping[str, Any], + project_id: Any, + iid: Any, + occurred_at: Optional[str], +) -> str: + event_marker = occurred_at + if event_marker is None: + event_marker = _first_value(payload, "webhook_id", "id") + if event_marker is None: + event_marker = _first_value(attributes, "webhook_id", "id") + if event_marker is None: + event_marker = "unknown" + return f"gitlab:{MERGE_WEBHOOK_ACTION}:{project_id}:{iid}:{event_marker}" + + +def _event_idempotency_key(event) -> Optional[str]: + """Prefer a transport id when an adapter makes webhook headers available.""" + + for attribute in ("webhook_id", "idempotency_key"): + value = getattr(event, attribute, None) + if value is not None and value != "": + return f"gitlab:merge: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:merge:header:{value}" + + data = getattr(event, "data", None) + if isinstance(data, Mapping): + for key in ("webhook_id", "idempotency_key"): + value = data.get(key) + if value is not None and value != "": + return f"gitlab:merge:header:{value}" + return None + + +def build_merged_notification(data: Mapping[str, Any]) -> MergeRequestNotification: + """Convert a completed GitLab MR merge webhook 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") + + webhook_action = _require_value(attributes, "action") + if webhook_action != MERGE_WEBHOOK_ACTION: + raise ValueError(f"unsupported merge action: {webhook_action}") + if attributes.get("state") != "merged" and not _first_value(attributes, "merged_at"): + raise ValueError("merge webhook does not describe a completed merge") + + project_id = _require_value(project_data, "id") + iid = _require_value(attributes, "iid") + 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") + occurred_at = _first_value(attributes, "merged_at", "actioned_at", "updated_at") + actor = _merge_actor(payload, attributes) + triggered_by_data = payload.get("user") + triggered_by = _normalize_actor(triggered_by_data) if isinstance(triggered_by_data, Mapping) else None + project_label = str(project_path) if project_path is not None else "项目不可用" + title_text = str(title) if title is not None and title != "" else "标题不可用" + mr_url_text = str(mr_url) if mr_url is not None else "MR 链接不可用" + occurred_at_text = str(occurred_at) if occurred_at is not None else "时间不可用" + actor_name = _actor_display_name(actor) + + message = ( + f"MR !{iid}「{title_text}」已合并: {actor_name} 合并了项目 {project_label}, " + f"时间: {occurred_at_text}, 链接: {mr_url_text}" + ) + return MergeRequestNotification( + source="gitlab", + event_type=MERGE_NOTIFICATION_EVENT_TYPE, + action=MERGED_NOTIFICATION_ACTION, + webhook_action=webhook_action, + message=message, + project={ + "id": project_id, + "path": project_path, + "url": project_url, + }, + merge_request={ + "iid": iid, + "title": title, + "url": mr_url, + }, + actor=actor, + occurred_at=occurred_at, + triggered_by=triggered_by, + idempotency_key=_merge_idempotency_key(payload, attributes, project_id, iid, occurred_at), + ) + + +class MergeRequestNotificationHooks: + """Handle completed MR merge webhooks without invoking the GitLab API.""" + + 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: + 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 != MERGE_WEBHOOK_ACTION: + self.logger.debug("Skip merged notification for action=%r", action) + return + notification = build_merged_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 merged 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: + self.logger.error( + "merged notification channel failed (project=%s, mr_iid=%s): %s", + notification.project.get("path") or notification.project.get("id"), + notification.merge_request.get("iid"), + 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/src/merge_request_hook.py b/src/hooks/merge_request.py similarity index 100% rename from src/merge_request_hook.py rename to src/hooks/merge_request.py diff --git a/src/note_hook.py b/src/hooks/note.py similarity index 100% rename from src/note_hook.py rename to src/hooks/note.py diff --git a/src/locales/en/LC_MESSAGES/gitlab-bot.po b/src/locales/en/LC_MESSAGES/gitlab-bot.po index 4dde89a..6e01232 100644 --- a/src/locales/en/LC_MESSAGES/gitlab-bot.po +++ b/src/locales/en/LC_MESSAGES/gitlab-bot.po @@ -17,18 +17,18 @@ msgstr "" "Content-Type: text/plain; charset=UTF-8\n" "Content-Transfer-Encoding: 8bit\n" -#: merge_request_hook.py:16 +#: hooks/merge_request.py:16 msgid "commit_subject_max_length" msgstr "🙁 The subject line must not exceed {commit_subject_max_length} characters." -#: merge_request_hook.py:27 +#: hooks/merge_request.py:27 msgid "milestone_is_required" msgstr "🙁 Milestone is required." msgid "issue_num_is_required" msgstr "Merge requests can only be merged if the source branch is associated with an existing issue." -#: merge_request_hook.py:83 +#: hooks/merge_request.py:83 msgid "bot_review_success" msgstr "😊 Review validation success and approve the merge request." diff --git a/src/locales/zh/LC_MESSAGES/gitlab-bot.po b/src/locales/zh/LC_MESSAGES/gitlab-bot.po index ec5eb29..9628166 100644 --- a/src/locales/zh/LC_MESSAGES/gitlab-bot.po +++ b/src/locales/zh/LC_MESSAGES/gitlab-bot.po @@ -17,18 +17,18 @@ msgstr "" "Content-Type: text/plain; charset=UTF-8\n" "Content-Transfer-Encoding: 8bit\n" -#: merge_request_hook.py:16 +#: hooks/merge_request.py:16 msgid "commit_subject_max_length" msgstr "🙁 主题行不能超过 {commit_subject_max_length} 个字符" -#: merge_request_hook.py:27 +#: hooks/merge_request.py:27 msgid "milestone_is_required" msgstr "🙁 必须选择里程碑" msgid "issue_num_is_required" msgstr "🙁 只有当源分支与现有问题相关联时，才能合并合并请求" -#: merge_request_hook.py:83 +#: hooks/merge_request.py:83 msgid "bot_review_success" msgstr "😊合并请求验证成功，批准合并请求。" diff --git a/src/notifications/__init__.py b/src/notifications/__init__.py new file mode 100644 index 0000000..31103f9 --- /dev/null +++ b/src/notifications/__init__.py @@ -0,0 +1,17 @@ +# 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. + +from .model import MergeRequestNotification + +__all__ = ["MergeRequestNotification"] diff --git a/src/notifications/model.py b/src/notifications/model.py new file mode 100644 index 0000000..9a7175c --- /dev/null +++ b/src/notifications/model.py @@ -0,0 +1,34 @@ +# 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. + +from dataclasses import dataclass +from typing import Any, Dict, Mapping, Optional + + +@dataclass +class MergeRequestNotification: + """Normalized, channel-independent information about an MR 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 + triggered_by: Optional[Dict[str, Any]] = None + idempotency_key: Optional[str] = None diff --git a/tests/fixtures/merge_webhook.py b/tests/fixtures/merge_webhook.py new file mode 100644 index 0000000..ed20131 --- /dev/null +++ b/tests/fixtures/merge_webhook.py @@ -0,0 +1,49 @@ +"""Representative redacted GitLab merge request webhook payload.""" + +from copy import deepcopy + + +def make_merge_webhook( + *, + action="merge", + state="merged", + merged_at="2026-08-02T10:05:00Z", + username="webhook-trigger", + include_merge_user=True, +): + attributes = { + "action": action, + "state": state, + "merged_at": merged_at, + "merge_user_id": 42 if include_merge_user else None, + "id": 9001, + "iid": 12, + "title": "Add feature", + "url": "https://gitlab.example.com/group/project/-/merge_requests/12", + "updated_at": "2026-08-02T10:05:00Z", + } + if include_merge_user: + attributes["merge_user"] = {"id": 42, "name": "Merger", "username": "merger"} + + return { + "object_kind": "merge_request", + "event_type": "merge_request", + "id": "merge-event-9001", + "user": {"id": 7, "name": "Webhook Trigger", "username": username}, + 'project': { + "id": 76, + "name": "Project", + "path_with_namespace": "group/project", + "web_url": "https://gitlab.example.com/group/project", + }, + "object_attributes": attributes, + } + + +MERGE_WEBHOOK = make_merge_webhook() + + +def copy_merge_webhook(**kwargs): + if not kwargs: + return deepcopy(MERGE_WEBHOOK) + return make_merge_webhook(**kwargs) diff --git a/tests/test_approval_notification.py b/tests/test_approval_notification.py index 9d163a0..255a705 100644 --- a/tests/test_approval_notification.py +++ b/tests/test_approval_notification.py @@ -7,11 +7,8 @@ from gidgetlab.sansio import Event import gitlab_bot -from src.approval_notification import ( - ApprovalNotificationHooks, - LogChannel, - build_notification, -) +from src.channels.log import LogChannel +from src.hooks.approval_notification import ApprovalNotificationHooks, build_notification from tests.fixtures.approval_webhook import copy_webhook diff --git a/tests/test_delivery_modules.py b/tests/test_delivery_modules.py new file mode 100644 index 0000000..faa44bb --- /dev/null +++ b/tests/test_delivery_modules.py @@ -0,0 +1,9 @@ +from src.delivery.coordinator import NotificationDelivery +from src.delivery.idempotent_channel import DurableIdempotentChannel +from src.delivery.sqlite import NotificationDeliveryStore + + +def test_delivery_components_are_available_from_responsibility_modules(): + assert NotificationDeliveryStore.__module__ == "src.delivery.sqlite" + assert DurableIdempotentChannel.__module__ == "src.delivery.idempotent_channel" + assert NotificationDelivery.__module__ == "src.delivery.coordinator" diff --git a/tests/test_hook_modules.py b/tests/test_hook_modules.py new file mode 100644 index 0000000..dbf425e --- /dev/null +++ b/tests/test_hook_modules.py @@ -0,0 +1,9 @@ +from src.hooks.approval_notification import ApprovalNotificationHooks, build_notification +from src.hooks.merge_notification import MergeRequestNotificationHooks, build_merged_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 build_notification.__module__ == "src.hooks.approval_notification" + assert build_merged_notification.__module__ == "src.hooks.merge_notification" diff --git a/tests/test_merge_notification.py b/tests/test_merge_notification.py new file mode 100644 index 0000000..dd07c72 --- /dev/null +++ b/tests/test_merge_notification.py @@ -0,0 +1,134 @@ +import asyncio +import logging +from types import SimpleNamespace + +import pytest +from gidgetlab.sansio import Event + +import gitlab_bot +from src.hooks.merge_notification import MergeRequestNotificationHooks, build_merged_notification +from tests.fixtures.merge_webhook import copy_merge_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_merge_webhook()) + + +def test_build_merged_notification_normalizes_fields_and_message(): + notification = build_merged_notification(copy_merge_webhook()) + + assert notification.source == "gitlab" + assert notification.event_type == "merge_request_lifecycle" + assert notification.action == "merged" + assert notification.webhook_action == "merge" + assert notification.actor["username"] == "merger" + assert notification.triggered_by["username"] == "webhook-trigger" + assert notification.occurred_at == "2026-08-02T10:05:00Z" + assert notification.idempotency_key == "gitlab:merge:76:12:2026-08-02T10:05:00Z" + assert "MR !12" in notification.message + assert "Add feature" in notification.message + assert "group/project" in notification.message + assert "Merger" in notification.message + assert "已合并" in notification.message + assert "https://gitlab.example.com/group/project/-/merge_requests/12" in notification.message + + +@pytest.mark.parametrize( + ("action", "state", "merged_at"), + [("open", "merged", "2026-08-02T10:05:00Z"), ("merge", "opened", None), ("merged", "merged", None)], +) +def test_non_merge_or_uncompleted_events_are_skipped(action, state, merged_at): + channel = RecordingChannel() + hooks = MergeRequestNotificationHooks(channel) + + asyncio.run( + hooks.handle( + make_event(copy_merge_webhook(action=action, state=state, merged_at=merged_at)) + ) + ) + + assert channel.notifications == [] + + +@pytest.mark.parametrize( + "username", + ["webhook-trigger", "review-bot"], +) +def test_merge_route_sends_user_and_bot_events_without_global_filter(monkeypatch, username): + channel = RecordingChannel() + monkeypatch.setattr(gitlab_bot, "merge_request_notification_hooks", MergeRequestNotificationHooks(channel)) + monkeypatch.setattr(gitlab_bot, "bot_gitlab_username", "review-bot") + event = Event(copy_merge_webhook(username=username), event="Merge Request Hook") + + asyncio.run(gitlab_bot.bot.router.dispatch(event, None)) + + assert len(channel.notifications) == 1 + assert channel.notifications[0].action == "merged" + assert channel.notifications[0].webhook_action == "merge" + + +def test_auto_merge_falls_back_without_using_webhook_trigger_user(): + payload = copy_merge_webhook(include_merge_user=False) + channel = RecordingChannel() + + asyncio.run(MergeRequestNotificationHooks(channel).handle(make_event(payload))) + + notification = channel.notifications[0] + assert notification.actor["name"] == "GitLab 自动合并" + assert notification.actor["username"] == "gitlab-auto-merge" + assert notification.triggered_by["username"] == "webhook-trigger" + assert "GitLab 自动合并" in notification.message + assert "webhook-trigger" not in notification.message + + +def test_noncritical_fields_use_placeholders_and_are_still_sent(): + payload = copy_merge_webhook() + payload["project"].pop("path_with_namespace") + payload["project"].pop("name") + payload["project"].pop("web_url") + payload["object_attributes"]["title"] = None + payload["object_attributes"].pop("url") + payload["object_attributes"].pop("updated_at") + payload["object_attributes"]["merged_at"] = None + channel = RecordingChannel() + + asyncio.run(MergeRequestNotificationHooks(channel).handle(make_event(payload))) + + notification = channel.notifications[0] + assert "项目不可用" in notification.message + assert "标题不可用" in notification.message + assert "MR 链接不可用" in notification.message + assert "时间不可用" in notification.message + + +def test_missing_critical_fields_are_logged_and_skipped(caplog): + payload = copy_merge_webhook() + payload["project"].pop("id") + channel = RecordingChannel() + logger = logging.getLogger("test.merge_notification.invalid") + caplog.set_level(logging.ERROR, logger=logger.name) + + asyncio.run(MergeRequestNotificationHooks(channel, logger=logger).handle(make_event(payload))) + + assert channel.notifications == [] + assert "invalid merged webhook" in caplog.text + + +def test_merge_notification_does_not_call_gitlab_api(): + channel = RecordingChannel() + api = SimpleNamespace( + getitem=lambda *_args, **_kwargs: pytest.fail("merge notification must not call GitLab API"), + post=lambda *_args, **_kwargs: pytest.fail("merge notification must not call GitLab API"), + ) + + asyncio.run(MergeRequestNotificationHooks(channel).handle(make_event(), api)) + + assert len(channel.notifications) == 1 diff --git a/tests/test_merge_request_approval_contract.py b/tests/test_merge_request_approval_contract.py index ff4fd96..5d244ef 100644 --- a/tests/test_merge_request_approval_contract.py +++ b/tests/test_merge_request_approval_contract.py @@ -1,6 +1,6 @@ import asyncio -import src.merge_request_hook as merge_request_hook +import src.hooks.merge_request as merge_request_hook from tests.fixtures.approval_contract import ( APPROVALS_EMPTY_GET, APPROVALS_OTHER_USER_GET, diff --git a/tests/test_merge_request_check_commit.py b/tests/test_merge_request_check_commit.py index a83126e..3ebf625 100644 --- a/tests/test_merge_request_check_commit.py +++ b/tests/test_merge_request_check_commit.py @@ -1,7 +1,7 @@ import asyncio from types import SimpleNamespace -import src.merge_request_hook as merge_request_hook +import src.hooks.merge_request as merge_request_hook from tests.fixtures.approval_contract import ( APPROVALS_ROBOT_GET, APPROVE_RESPONSE, diff --git a/tests/test_merge_request_failure_state.py b/tests/test_merge_request_failure_state.py index 0ae77e9..8cccfcc 100644 --- a/tests/test_merge_request_failure_state.py +++ b/tests/test_merge_request_failure_state.py @@ -1,7 +1,7 @@ import asyncio from types import SimpleNamespace -import src.merge_request_hook as merge_request_hook +import src.hooks.merge_request as merge_request_hook from tests.fixtures.approval_contract import ( APPROVALS_EMPTY_GET, APPROVALS_OTHER_USER_GET, diff --git a/tests/test_merge_request_webhook_regression.py b/tests/test_merge_request_webhook_regression.py index ccd8623..ddd6884 100644 --- a/tests/test_merge_request_webhook_regression.py +++ b/tests/test_merge_request_webhook_regression.py @@ -3,7 +3,7 @@ import pytest -import src.merge_request_hook as merge_request_hook +import src.hooks.merge_request as merge_request_hook from tests.fixtures.approval_contract import ( APPROVALS_EMPTY_GET, APPROVALS_ROBOT_GET, diff --git a/tests/test_notification_contracts.py b/tests/test_notification_contracts.py new file mode 100644 index 0000000..c9e8360 --- /dev/null +++ b/tests/test_notification_contracts.py @@ -0,0 +1,18 @@ +from src.channels.base import Channel +from src.channels.log import LogChannel +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 + + +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" + + +def test_delivery_components_are_available_from_responsibility_modules(): + assert NotificationDeliveryStore.__module__ == "src.delivery.sqlite" + assert DurableIdempotentChannel.__module__ == "src.delivery.idempotent_channel" + assert NotificationDelivery.__module__ == "src.delivery.coordinator" diff --git a/tests/test_notification_delivery.py b/tests/test_notification_delivery.py new file mode 100644 index 0000000..ed35426 --- /dev/null +++ b/tests/test_notification_delivery.py @@ -0,0 +1,208 @@ +import asyncio +import json +import logging +from types import SimpleNamespace + +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 MergeRequestNotificationHooks, build_merged_notification +from tests.fixtures.merge_webhook import copy_merge_webhook + + +class RecordingChannel: + def __init__(self): + self.notifications = [] + self.calls = 0 + + async def send(self, notification): + self.calls += 1 + self.notifications.append(notification) + + +class YieldingChannel(RecordingChannel): + async def send(self, notification): + self.calls += 1 + await asyncio.sleep(0.01) + 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 channel failure") + self.notifications.append(notification) + + +class CrashAfterChannel: + def __init__(self, channel): + self.channel = channel + + async def send(self, notification): + await self.channel.send(notification) + raise KeyboardInterrupt + + +def make_notification(): + return build_merged_notification(copy_merge_webhook()) + + +def make_delivery(tmp_path, channel, *, timeout=300): + store = NotificationDeliveryStore(str(tmp_path / "delivery.sqlite3"), sending_timeout_seconds=timeout) + durable_channel = DurableIdempotentChannel(channel, store) + return store, NotificationDelivery(durable_channel, store) + + +def test_duplicate_delivery_has_one_channel_effect(tmp_path): + channel = RecordingChannel() + store, delivery = make_delivery(tmp_path, channel) + notification = make_notification() + + async def run(): + first = await delivery.deliver(notification) + second = await delivery.deliver(notification) + return first, second + + first, second = asyncio.run(run()) + + assert first is True + assert second is False + assert channel.calls == 1 + assert store.get_delivery(notification.idempotency_key).status == "sent" + assert store.get_channel_effect(notification.idempotency_key)["status"] == "accepted" + + +def test_concurrent_delivery_has_one_channel_effect(tmp_path): + channel = YieldingChannel() + store, delivery = make_delivery(tmp_path, channel) + notification = make_notification() + + async def run(): + return await asyncio.gather( + delivery.deliver(notification), + delivery.deliver(notification), + ) + + results = asyncio.run(run()) + + assert sorted(results) == [False, True] + assert channel.calls == 1 + assert store.get_delivery(notification.idempotency_key).status == "sent" + + +def test_channel_failure_is_retryable_and_reuses_key(tmp_path): + channel = FailingOnceChannel() + store, delivery = make_delivery(tmp_path, channel) + notification = make_notification() + + async def run(): + first = await delivery.deliver(notification) + failed = store.get_delivery(notification.idempotency_key) + second = await delivery.deliver(notification) + return first, failed, second + + first, failed, second = asyncio.run(run()) + + assert first is False + assert failed.status == "failed" + assert second is True + assert channel.calls == 2 + assert store.get_delivery(notification.idempotency_key).status == "sent" + assert store.get_delivery(notification.idempotency_key).attempts == 2 + + +def test_restart_recovers_failed_delivery_and_manual_replay(tmp_path): + first_channel = FailingOnceChannel() + store, first_delivery = make_delivery(tmp_path, first_channel) + notification = make_notification() + asyncio.run(first_delivery.deliver(notification)) + + second_channel = RecordingChannel() + second_store = NotificationDeliveryStore(str(tmp_path / "delivery.sqlite3")) + second_durable_channel = DurableIdempotentChannel(second_channel, second_store) + second_delivery = NotificationDelivery(second_durable_channel, second_store) + + replayed = asyncio.run(second_delivery.replay_failed()) + + assert replayed == 1 + assert second_channel.calls == 1 + assert second_store.get_delivery(notification.idempotency_key).status == "sent" + + +def test_restart_recovers_stale_sending_delivery(tmp_path): + channel = RecordingChannel() + store = NotificationDeliveryStore(str(tmp_path / "delivery.sqlite3"), sending_timeout_seconds=0) + notification = make_notification() + store.begin_delivery(notification) + + delivery = NotificationDelivery(DurableIdempotentChannel(channel, store), store) + recovered = asyncio.run(delivery.recover()) + + assert recovered == 1 + assert channel.calls == 1 + assert store.get_delivery(notification.idempotency_key).status == "sent" + + +def test_crash_after_channel_acceptance_does_not_duplicate_effect(tmp_path): + channel = RecordingChannel() + store = NotificationDeliveryStore(str(tmp_path / "delivery.sqlite3"), sending_timeout_seconds=0) + crashing_delivery = NotificationDelivery( + DurableIdempotentChannel(CrashAfterChannel(channel), store), + store, + ) + notification = make_notification() + + try: + asyncio.run(crashing_delivery.deliver(notification)) + except KeyboardInterrupt: + pass + + assert channel.calls == 1 + assert store.get_delivery(notification.idempotency_key).status == "sending" + assert store.get_channel_effect(notification.idempotency_key)["status"] == "reserved" + + recovered = NotificationDelivery( + DurableIdempotentChannel(channel, store), + store, + ) + assert asyncio.run(recovered.recover()) == 1 + assert channel.calls == 1 + assert store.get_delivery(notification.idempotency_key).status == "sent" + assert store.get_channel_effect(notification.idempotency_key)["status"] == "accepted" + + +def test_transport_idempotency_header_overrides_fallback_key(tmp_path): + channel = RecordingChannel() + store, delivery = make_delivery(tmp_path, channel) + hooks = MergeRequestNotificationHooks( + DurableIdempotentChannel(channel, store), + delivery=delivery, + ) + event = SimpleNamespace(data=copy_merge_webhook(), headers={"webhook-id": "hook-123"}) + + asyncio.run(hooks.handle(event)) + + notification = channel.notifications[0] + assert notification.idempotency_key == "gitlab:merge:header:hook-123" + assert store.get_delivery(notification.idempotency_key).status == "sent" + + +def test_delivery_emits_structured_observability(caplog, tmp_path): + logger = logging.getLogger("test.notification_delivery") + caplog.set_level(logging.INFO, logger=logger.name) + channel = RecordingChannel() + store = NotificationDeliveryStore(str(tmp_path / "delivery.sqlite3")) + delivery = NotificationDelivery(DurableIdempotentChannel(channel, store), store, logger=logger) + + asyncio.run(delivery.deliver(make_notification())) + + payload = json.loads(caplog.records[-1].message) + assert payload["event"] == "merge_notification_delivery" + assert payload["action"] == "sent" + assert delivery.counters["sent"] == 1 diff --git a/tests/test_notification_module_boundaries.py b/tests/test_notification_module_boundaries.py new file mode 100644 index 0000000..d1e372a --- /dev/null +++ b/tests/test_notification_module_boundaries.py @@ -0,0 +1,104 @@ +import ast +import asyncio +from pathlib import Path + +from gidgetlab.sansio import Event + +import gitlab_bot +from src.hooks.approval_notification import ApprovalNotificationHooks +from src.hooks.merge_notification import MergeRequestNotificationHooks +from tests.fixtures.approval_webhook import copy_webhook +from tests.fixtures.merge_webhook import copy_merge_webhook + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +SRC_ROOT = PROJECT_ROOT / "src" + + +def _import_modules(path: Path): + tree = ast.parse(path.read_text()) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + yield from (alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + prefix = "." * node.level + yield prefix + (node.module or "") + + +def test_channels_do_not_depend_on_hooks_or_delivery(): + for path in (SRC_ROOT / "channels").glob("*.py"): + imports = tuple(_import_modules(path)) + assert not any("hooks" in name or "delivery" in name for name in imports), path + + +def test_notification_hooks_only_depend_on_notification_model_and_channel_contract(): + notification_hook_paths = ( + SRC_ROOT / "hooks" / "approval_notification.py", + SRC_ROOT / "hooks" / "merge_notification.py", + ) + for path in notification_hook_paths: + imports = tuple(_import_modules(path)) + assert not any("channels.log" in name or "delivery" in name or "gitlab_bot" in name for name in imports), path + + +def test_delivery_does_not_depend_on_approval_notification_or_hooks(): + for path in (SRC_ROOT / "delivery").glob("*.py"): + imports = tuple(_import_modules(path)) + assert not any("approval_notification" in name or "hooks" in name for name in imports), path + + +def test_legacy_facades_are_removed(): + legacy_paths = ( + SRC_ROOT / "approval_notification.py", + SRC_ROOT / "notification_delivery.py", + SRC_ROOT / "issue_hook.py", + SRC_ROOT / "merge_request_hook.py", + SRC_ROOT / "note_hook.py", + ) + for path in legacy_paths: + assert not path.exists(), path + + +def test_legacy_module_imports_are_removed_from_source_and_tests(): + legacy_modules = { + "src.approval_notification", + "src.notification_delivery", + "src.issue_hook", + "src.merge_request_hook", + "src.note_hook", + } + python_paths = tuple(SRC_ROOT.rglob("*.py")) + tuple((PROJECT_ROOT / "tests").rglob("*.py")) + for path in python_paths: + imports = tuple(_import_modules(path)) + assert not any(name in legacy_modules for name in imports), path + + +def test_startup_recovery_and_notification_routes_are_registered_once(monkeypatch): + assert gitlab_bot.bot.app.on_startup.count(gitlab_bot.recover_merge_notification_deliveries) == 1 + + class RecordingChannel: + def __init__(self): + self.notifications = [] + + async def send(self, notification): + self.notifications.append(notification) + + approval_channel = RecordingChannel() + monkeypatch.setattr(gitlab_bot, "approval_notification_hooks", ApprovalNotificationHooks(approval_channel)) + monkeypatch.setattr(gitlab_bot, "bot_gitlab_username", "review-bot") + asyncio.run( + gitlab_bot.bot.router.dispatch( + Event(copy_webhook(username="reviewer"), event="Merge Request Hook"), + None, + ) + ) + assert len(approval_channel.notifications) == 1 + + merge_channel = RecordingChannel() + monkeypatch.setattr(gitlab_bot, "merge_request_notification_hooks", MergeRequestNotificationHooks(merge_channel)) + asyncio.run( + gitlab_bot.bot.router.dispatch( + Event(copy_merge_webhook(username="merger"), event="Merge Request Hook"), + None, + ) + ) + assert len(merge_channel.notifications) == 1 diff --git a/tests/test_runtime_startup.py b/tests/test_runtime_startup.py index d09bc43..cce68f8 100644 --- a/tests/test_runtime_startup.py +++ b/tests/test_runtime_startup.py @@ -2,3 +2,9 @@ def test_gidgetlab_runtime_import_is_available(): from gitlab_bot import GitLabBot assert GitLabBot is not None + + +def test_merge_notification_recovery_is_registered_on_startup(): + from gitlab_bot import bot, recover_merge_notification_deliveries + + assert recover_merge_notification_deliveries in bot.app.on_startup diff --git a/tests/test_webhook_hook_modules.py b/tests/test_webhook_hook_modules.py new file mode 100644 index 0000000..fe4534a --- /dev/null +++ b/tests/test_webhook_hook_modules.py @@ -0,0 +1,38 @@ +import ast +from pathlib import Path + +import gitlab_bot + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +SRC_ROOT = PROJECT_ROOT / "src" + + +def test_webhook_implementations_live_in_hooks_package(): + assert gitlab_bot.issue_hooks.__class__.__module__ == "src.hooks.issue" + assert gitlab_bot.merge_request_hooks.__class__.__module__ == "src.hooks.merge_request" + assert gitlab_bot.note_hooks.__class__.__module__ == "src.hooks.note" + + +def test_legacy_webhook_modules_are_removed(): + for path in ( + SRC_ROOT / "issue_hook.py", + SRC_ROOT / "merge_request_hook.py", + SRC_ROOT / "note_hook.py", + ): + assert not path.exists(), path + + +def test_new_webhook_modules_do_not_import_legacy_paths(): + for path in ( + SRC_ROOT / "hooks" / "issue.py", + SRC_ROOT / "hooks" / "merge_request.py", + SRC_ROOT / "hooks" / "note.py", + ): + tree = ast.parse(path.read_text()) + imports = [] + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imports.extend(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + imports.append(("." * node.level) + (node.module or "")) + assert not any(name in {"src.issue_hook", "src.merge_request_hook", "src.note_hook"} for name in imports), path