Add mention decorator for GitHub command handling

Author: joshuadavidthomas

src/django_github_app/commands.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import NamedTuple
+
+
+class EventAction(NamedTuple):
+    event: str
+    action: str
+
+
+class CommandScope(str, Enum):
+    COMMIT = "commit"
+    ISSUE = "issue"
+    PR = "pr"
+
+    def get_events(self) -> list[EventAction]:
+        match self:
+            case CommandScope.ISSUE:
+                return [
+                    EventAction("issue_comment", "created"),
+                ]
+            case CommandScope.PR:
+                return [
+                    EventAction("issue_comment", "created"),
+                    EventAction("pull_request_review_comment", "created"),
+                    EventAction("pull_request_review", "submitted"),
+                ]
+            case CommandScope.COMMIT:
+                return [
+                    EventAction("commit_comment", "created"),
+                ]
+
+    @classmethod
+    def all_events(cls) -> list[EventAction]:
+        return list(
+            dict.fromkeys(
+                event_action for scope in cls for event_action in scope.get_events()
+            )
+        )
+

src/django_github_app/routing.py

@@ -1,36 +1,111 @@
 from __future__ import annotations
 
+from asyncio import iscoroutinefunction
 from collections.abc import Awaitable
 from collections.abc import Callable
+from functools import wraps
 from typing import Any
+from typing import Protocol
 from typing import TypeVar
+from typing import cast
 
 from django.utils.functional import classproperty
 from gidgethub import sansio
 from gidgethub.routing import Router as GidgetHubRouter
 
 from ._typing import override
+from .commands import CommandScope
 
 AsyncCallback = Callable[..., Awaitable[None]]
 SyncCallback = Callable[..., None]
 
 CB = TypeVar("CB", AsyncCallback, SyncCallback)
 
 
+class MentionHandlerBase(Protocol):
+    _mention_command: str | None
+    _mention_scope: CommandScope | None
+    _mention_permission: str | None
+
+
+class AsyncMentionHandler(MentionHandlerBase, Protocol):
+    async def __call__(
+        self, event: sansio.Event, *args: Any, **kwargs: Any
+    ) -> None: ...
+
+
+class SyncMentionHandler(MentionHandlerBase, Protocol):
+    def __call__(self, event: sansio.Event, *args: Any, **kwargs: Any) -> None: ...
+
+
+MentionHandler = AsyncMentionHandler | SyncMentionHandler
+
+
 class GitHubRouter(GidgetHubRouter):
     _routers: list[GidgetHubRouter] = []
 
     def __init__(self, *args) -> None:
         super().__init__(*args)
         GitHubRouter._routers.append(self)
 
+    @override
+    def add(
+        self, func: AsyncCallback | SyncCallback, event_type: str, **data_detail: Any
+    ) -> None:
+        """Override to accept both async and sync callbacks."""
+        super().add(cast(AsyncCallback, func), event_type, **data_detail)
+
     @classproperty
     def routers(cls):
         return list(cls._routers)
 
     def event(self, event_type: str, **kwargs: Any) -> Callable[[CB], CB]:
         def decorator(func: CB) -> CB:
-            self.add(func, event_type, **kwargs)  # type: ignore[arg-type]
+            self.add(func, event_type, **kwargs)
+            return func
+
+        return decorator
+
+    def mention(self, **kwargs: Any) -> Callable[[CB], CB]:
+        def decorator(func: CB) -> CB:
+            command = kwargs.pop("command", None)
+            scope = kwargs.pop("scope", None)
+            permission = kwargs.pop("permission", None)
+
+            @wraps(func)
+            async def async_wrapper(
+                event: sansio.Event, *args: Any, **wrapper_kwargs: Any
+            ) -> None:
+                # TODO: Parse comment body for mentions
+                # TODO: If command specified, check if it matches
+                # TODO: Check permissions
+                # For now, just call through
+                await func(event, *args, **wrapper_kwargs)  # type: ignore[func-returns-value]
+
+            @wraps(func)
+            def sync_wrapper(
+                event: sansio.Event, *args: Any, **wrapper_kwargs: Any
+            ) -> None:
+                # TODO: Parse comment body for mentions
+                # TODO: If command specified, check if it matches
+                # TODO: Check permissions
+                # For now, just call through
+                func(event, *args, **wrapper_kwargs)
+
+            wrapper: MentionHandler
+            if iscoroutinefunction(func):
+                wrapper = cast(AsyncMentionHandler, async_wrapper)
+            else:
+                wrapper = cast(SyncMentionHandler, sync_wrapper)
+
+            wrapper._mention_command = command.lower() if command else None
+            wrapper._mention_scope = scope
+            wrapper._mention_permission = permission
+
+            events = scope.get_events() if scope else CommandScope.all_events()
+            for event_action in events:
+                self.add(wrapper, event_action.event, action=event_action.action, **kwargs)
+
             return func
 
         return decorator

tests/test_routing.py

@@ -1,9 +1,13 @@
 from __future__ import annotations
 
+import asyncio
+
 import pytest
 from django.http import HttpRequest
 from django.http import JsonResponse
+from gidgethub import sansio
 
+from django_github_app.commands import CommandScope
 from django_github_app.github import SyncGitHubAPI
 from django_github_app.routing import GitHubRouter
 from django_github_app.views import BaseWebhookView
@@ -109,3 +113,174 @@ def test_router_memory_stress_test_legacy(self):
 
         assert len(views) == view_count
         assert not all(view.router is view1_router for view in views)
+
+
+class TestMentionDecorator:
+    def test_basic_mention_no_command(self, test_router):
+        handler_called = False
+        handler_args = None
+
+        @test_router.mention()
+        def handle_mention(event, *args, **kwargs):
+            nonlocal handler_called, handler_args
+            handler_called = True
+            handler_args = (event, args, kwargs)
+
+        event = sansio.Event(
+            {"action": "created", "comment": {"body": "@bot hello"}},
+            event="issue_comment",
+            delivery_id="123",
+        )
+        test_router.dispatch(event, None)
+
+        assert handler_called
+        assert handler_args[0] == event
+
+    def test_mention_with_command(self, test_router):
+        handler_called = False
+
+        @test_router.mention(command="help")
+        def help_command(event, *args, **kwargs):
+            nonlocal handler_called
+            handler_called = True
+            return "help response"
+
+        event = sansio.Event(
+            {"action": "created", "comment": {"body": "@bot help"}},
+            event="issue_comment",
+            delivery_id="123",
+        )
+        test_router.dispatch(event, None)
+
+        assert handler_called
+
+    def test_mention_with_scope(self, test_router):
+        pr_handler_called = False
+
+        @test_router.mention(command="deploy", scope=CommandScope.PR)
+        def deploy_command(event, *args, **kwargs):
+            nonlocal pr_handler_called
+            pr_handler_called = True
+
+        pr_event = sansio.Event(
+            {"action": "created", "comment": {"body": "@bot deploy"}},
+            event="pull_request_review_comment",
+            delivery_id="123",
+        )
+        test_router.dispatch(pr_event, None)
+
+        assert pr_handler_called
+
+        issue_event = sansio.Event(
+            {"action": "created", "comment": {"body": "@bot deploy"}},
+            event="commit_comment",  # This is NOT a PR event
+            delivery_id="124",
+        )
+        pr_handler_called = False  # Reset
+
+        test_router.dispatch(issue_event, None)
+
+        assert not pr_handler_called
+
+    def test_mention_with_permission(self, test_router):
+        handler_called = False
+
+        @test_router.mention(command="delete", permission="admin")
+        def delete_command(event, *args, **kwargs):
+            nonlocal handler_called
+            handler_called = True
+
+        event = sansio.Event(
+            {"action": "created", "comment": {"body": "@bot delete"}},
+            event="issue_comment",
+            delivery_id="123",
+        )
+        test_router.dispatch(event, None)
+
+        assert handler_called
+
+    def test_case_insensitive_command(self, test_router):
+        handler_called = False
+
+        @test_router.mention(command="HELP")
+        def help_command(event, *args, **kwargs):
+            nonlocal handler_called
+            handler_called = True
+
+        event = sansio.Event(
+            {"action": "created", "comment": {"body": "@bot help"}},
+            event="issue_comment",
+            delivery_id="123",
+        )
+        test_router.dispatch(event, None)
+
+        assert handler_called
+
+    def test_multiple_decorators_on_same_function(self, test_router):
+        call_count = 0
+
+        @test_router.mention(command="help")
+        @test_router.mention(command="h")
+        @test_router.mention(command="?")
+        def help_command(event, *args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            return f"help called {call_count} times"
+
+        event1 = sansio.Event(
+            {"action": "created", "comment": {"body": "@bot help"}},
+            event="issue_comment",
+            delivery_id="123",
+        )
+        test_router.dispatch(event1, None)
+
+        assert call_count == 3
+
+        call_count = 0
+        event2 = sansio.Event(
+            {"action": "created", "comment": {"body": "@bot h"}},
+            event="issue_comment",
+            delivery_id="124",
+        )
+        test_router.dispatch(event2, None)
+
+        assert call_count == 3
+
+        # This behavior will change once we implement command parsing
+
+    def test_async_mention_handler(self, test_router):
+        handler_called = False
+
+        @test_router.mention(command="async-test")
+        async def async_handler(event, *args, **kwargs):
+            nonlocal handler_called
+            handler_called = True
+            return "async response"
+
+        event = sansio.Event(
+            {"action": "created", "comment": {"body": "@bot async-test"}},
+            event="issue_comment",
+            delivery_id="123",
+        )
+
+        asyncio.run(test_router.adispatch(event, None))
+
+        assert handler_called
+
+    def test_sync_mention_handler(self, test_router):
+        handler_called = False
+
+        @test_router.mention(command="sync-test")
+        def sync_handler(event, *args, **kwargs):
+            nonlocal handler_called
+            handler_called = True
+            return "sync response"
+
+        event = sansio.Event(
+            {"action": "created", "comment": {"body": "@bot sync-test"}},
+            event="issue_comment",
+            delivery_id="123",
+        )
+        test_router.dispatch(event, None)
+
+        assert handler_called