Simplify permission checking and remove optional return types

Author: joshuadavidthomas

src/django_github_app/mentions.py

@@ -121,7 +121,7 @@ def from_event(cls, event: sansio.Event) -> Comment:
 class MentionContext:
     comment: Comment
     triggered_by: Mention
-    user_permission: Permission | None
+    user_permission: Permission
     scope: MentionScope | None
 
 

src/django_github_app/permissions.py

@@ -48,10 +48,33 @@ class PermissionCacheKey(NamedTuple):
     username: str
 
 
-async def aget_user_permission(
-    gh: AsyncGitHubAPI, owner: str, repo: str, username: str
+class EventInfo(NamedTuple):
+    author: str | None
+    owner: str | None
+    repo: str | None
+
+    @classmethod
+    def from_event(cls, event: sansio.Event) -> EventInfo:
+        comment = event.data.get("comment", {})
+        repository = event.data.get("repository", {})
+        
+        author = comment.get("user", {}).get("login")
+        owner = repository.get("owner", {}).get("login")
+        repo = repository.get("name")
+
+        return cls(author=author, owner=owner, repo=repo)
+
+
+async def aget_user_permission_from_event(
+    event: sansio.Event, gh: AsyncGitHubAPI
 ) -> Permission:
-    cache_key = PermissionCacheKey(owner, repo, username)
+    author, owner, repo = EventInfo.from_event(event)
+
+    if not (author and owner and repo):
+        return Permission.NONE
+
+    # Inline the logic from aget_user_permission
+    cache_key = PermissionCacheKey(owner, repo, author)
 
     if cache_key in cache:
         return cache[cache_key]
@@ -61,7 +84,7 @@ async def aget_user_permission(
     try:
         # Check if user is a collaborator and get their permission
         data = await gh.getitem(
-            f"/repos/{owner}/{repo}/collaborators/{username}/permission"
+            f"/repos/{owner}/{repo}/collaborators/{author}/permission"
         )
         permission_str = data.get("permission", "none")
         permission = Permission.from_string(permission_str)
@@ -80,10 +103,16 @@ async def aget_user_permission(
     return permission
 
 
-def get_user_permission(
-    gh: SyncGitHubAPI, owner: str, repo: str, username: str
+def get_user_permission_from_event(
+    event: sansio.Event, gh: SyncGitHubAPI
 ) -> Permission:
-    cache_key = PermissionCacheKey(owner, repo, username)
+    author, owner, repo = EventInfo.from_event(event)
+
+    if not (author and owner and repo):
+        return Permission.NONE
+
+    # Inline the logic from get_user_permission
+    cache_key = PermissionCacheKey(owner, repo, author)
 
     if cache_key in cache:
         return cache[cache_key]
@@ -92,7 +121,7 @@ def get_user_permission(
 
     try:
         # Check if user is a collaborator and get their permission
-        data = gh.getitem(f"/repos/{owner}/{repo}/collaborators/{username}/permission")
+        data = gh.getitem(f"/repos/{owner}/{repo}/collaborators/{author}/permission")
         permission_str = data.get("permission", "none")  # type: ignore[attr-defined]
         permission = Permission.from_string(permission_str)
     except gidgethub.HTTPException as e:
@@ -108,72 +137,3 @@ def get_user_permission(
 
     cache[cache_key] = permission
     return permission
-
-
-class EventInfo(NamedTuple):
-    comment_author: str | None
-    owner: str | None
-    repo: str | None
-
-    @classmethod
-    def from_event(cls, event: sansio.Event) -> EventInfo:
-        comment_author = None
-        owner = None
-        repo = None
-
-        if "comment" in event.data:
-            comment_author = event.data["comment"]["user"]["login"]
-
-        if "repository" in event.data:
-            owner = event.data["repository"]["owner"]["login"]
-            repo = event.data["repository"]["name"]
-
-        return cls(comment_author=comment_author, owner=owner, repo=repo)
-
-
-class PermissionCheck(NamedTuple):
-    has_permission: bool
-
-
-async def aget_user_permission_from_event(
-    event: sansio.Event, gh: AsyncGitHubAPI
-) -> Permission | None:
-    comment_author, owner, repo = EventInfo.from_event(event)
-
-    if not (comment_author and owner and repo):
-        return None
-
-    return await aget_user_permission(gh, owner, repo, comment_author)
-
-
-async def acheck_mention_permission(
-    event: sansio.Event, gh: AsyncGitHubAPI, required_permission: Permission
-) -> PermissionCheck:
-    user_permission = await aget_user_permission_from_event(event, gh)
-
-    if user_permission is None:
-        return PermissionCheck(has_permission=False)
-
-    return PermissionCheck(has_permission=user_permission >= required_permission)
-
-
-def get_user_permission_from_event(
-    event: sansio.Event, gh: SyncGitHubAPI
-) -> Permission | None:
-    comment_author, owner, repo = EventInfo.from_event(event)
-
-    if not (comment_author and owner and repo):
-        return None
-
-    return get_user_permission(gh, owner, repo, comment_author)
-
-
-def check_mention_permission(
-    event: sansio.Event, gh: SyncGitHubAPI, required_permission: Permission
-) -> PermissionCheck:
-    user_permission = get_user_permission_from_event(event, gh)
-
-    if user_permission is None:
-        return PermissionCheck(has_permission=False)
-
-    return PermissionCheck(has_permission=user_permission >= required_permission)

src/django_github_app/routing.py

@@ -98,11 +98,13 @@ async def async_wrapper(
                 if not mentions:
                     return
 
-                user_permission = await aget_user_permission_from_event(event, gh)
                 if permission is not None:
+                    user_permission = await aget_user_permission_from_event(event, gh)
                     required_perm = Permission[permission.upper()]
-                    if user_permission is None or user_permission < required_perm:
+                    if user_permission < required_perm:
                         return
+                else:
+                    user_permission = Permission.NONE
 
                 comment = Comment.from_event(event)
                 comment.mentions = mentions
@@ -135,11 +137,13 @@ def sync_wrapper(
                 if not mentions:
                     return
 
-                user_permission = get_user_permission_from_event(event, gh)
                 if permission is not None:
+                    user_permission = get_user_permission_from_event(event, gh)
                     required_perm = Permission[permission.upper()]
-                    if user_permission is None or user_permission < required_perm:
+                    if user_permission < required_perm:
                         return
+                else:
+                    user_permission = Permission.NONE
 
                 comment = Comment.from_event(event)
                 comment.mentions = mentions

tests/test_permissions.py

@@ -6,13 +6,14 @@
 
 import gidgethub
 import pytest
+from gidgethub import sansio
 
 from django_github_app.github import AsyncGitHubAPI
 from django_github_app.github import SyncGitHubAPI
 from django_github_app.permissions import Permission
-from django_github_app.permissions import aget_user_permission
+from django_github_app.permissions import aget_user_permission_from_event
 from django_github_app.permissions import cache
-from django_github_app.permissions import get_user_permission
+from django_github_app.permissions import get_user_permission_from_event
 
 
 @pytest.fixture(autouse=True)
@@ -22,6 +23,18 @@ def clear_cache():
     cache.clear()
 
 
+def create_test_event(username: str, owner: str, repo: str) -> sansio.Event:
+    """Create a test event with comment author and repository info."""
+    return sansio.Event(
+        {
+            "comment": {"user": {"login": username}},
+            "repository": {"owner": {"login": owner}, "name": repo},
+        },
+        event="issue_comment",
+        delivery_id="test",
+    )
+
+
 class TestPermission:
     def test_permission_ordering(self):
         assert Permission.NONE < Permission.READ
@@ -64,8 +77,9 @@ class TestGetUserPermission:
     async def test_collaborator_with_admin_permission(self):
         gh = create_autospec(AsyncGitHubAPI, instance=True)
         gh.getitem = AsyncMock(return_value={"permission": "admin"})
+        event = create_test_event("user", "owner", "repo")
 
-        permission = await aget_user_permission(gh, "owner", "repo", "user")
+        permission = await aget_user_permission_from_event(event, gh)
 
         assert permission == Permission.ADMIN
         gh.getitem.assert_called_once_with(
@@ -75,8 +89,9 @@ async def test_collaborator_with_admin_permission(self):
     async def test_collaborator_with_write_permission(self):
         gh = create_autospec(AsyncGitHubAPI, instance=True)
         gh.getitem = AsyncMock(return_value={"permission": "write"})
+        event = create_test_event("user", "owner", "repo")
 
-        permission = await aget_user_permission(gh, "owner", "repo", "user")
+        permission = await aget_user_permission_from_event(event, gh)
 
         assert permission == Permission.WRITE
 
@@ -90,7 +105,8 @@ async def test_non_collaborator_public_repo(self):
             ]
         )
 
-        permission = await aget_user_permission(gh, "owner", "repo", "user")
+        event = create_test_event("user", "owner", "repo")
+        permission = await aget_user_permission_from_event(event, gh)
 
         assert permission == Permission.READ
         assert gh.getitem.call_count == 2
@@ -106,8 +122,9 @@ async def test_non_collaborator_private_repo(self):
                 {"private": True},  # Repo is private
             ]
         )
+        event = create_test_event("user", "owner", "repo")
 
-        permission = await aget_user_permission(gh, "owner", "repo", "user")
+        permission = await aget_user_permission_from_event(event, gh)
 
         assert permission == Permission.NONE
 
@@ -116,16 +133,18 @@ async def test_api_error_returns_none_permission(self):
         gh.getitem = AsyncMock(
             side_effect=gidgethub.HTTPException(500, "Server error", {})
         )
+        event = create_test_event("user", "owner", "repo")
 
-        permission = await aget_user_permission(gh, "owner", "repo", "user")
+        permission = await aget_user_permission_from_event(event, gh)
 
         assert permission == Permission.NONE
 
     async def test_missing_permission_field(self):
         gh = create_autospec(AsyncGitHubAPI, instance=True)
         gh.getitem = AsyncMock(return_value={})  # No permission field
+        event = create_test_event("user", "owner", "repo")
 
-        permission = await aget_user_permission(gh, "owner", "repo", "user")
+        permission = await aget_user_permission_from_event(event, gh)
 
         assert permission == Permission.NONE
 
@@ -134,8 +153,9 @@ class TestGetUserPermissionSync:
     def test_collaborator_with_permission(self):
         gh = create_autospec(SyncGitHubAPI, instance=True)
         gh.getitem = Mock(return_value={"permission": "maintain"})
+        event = create_test_event("user", "owner", "repo")
 
-        permission = get_user_permission(gh, "owner", "repo", "user")
+        permission = get_user_permission_from_event(event, gh)
 
         assert permission == Permission.MAINTAIN
         gh.getitem.assert_called_once_with(
@@ -151,8 +171,9 @@ def test_non_collaborator_public_repo(self):
                 {"private": False},  # Repo is public
             ]
         )
+        event = create_test_event("user", "owner", "repo")
 
-        permission = get_user_permission(gh, "owner", "repo", "user")
+        permission = get_user_permission_from_event(event, gh)
 
         assert permission == Permission.READ
 
@@ -162,14 +183,15 @@ class TestPermissionCaching:
     async def test_cache_hit(self):
         gh = create_autospec(AsyncGitHubAPI, instance=True)
         gh.getitem = AsyncMock(return_value={"permission": "write"})
+        event = create_test_event("user", "owner", "repo")
 
         # First call should hit the API
-        perm1 = await aget_user_permission(gh, "owner", "repo", "user")
+        perm1 = await aget_user_permission_from_event(event, gh)
         assert perm1 == Permission.WRITE
         assert gh.getitem.call_count == 1
 
         # Second call should use cache
-        perm2 = await aget_user_permission(gh, "owner", "repo", "user")
+        perm2 = await aget_user_permission_from_event(event, gh)
         assert perm2 == Permission.WRITE
         assert gh.getitem.call_count == 1  # No additional API call
 
@@ -182,9 +204,11 @@ async def test_cache_different_users(self):
                 {"permission": "admin"},
             ]
         )
+        event1 = create_test_event("user1", "owner", "repo")
+        event2 = create_test_event("user2", "owner", "repo")
 
-        perm1 = await aget_user_permission(gh, "owner", "repo", "user1")
-        perm2 = await aget_user_permission(gh, "owner", "repo", "user2")
+        perm1 = await aget_user_permission_from_event(event1, gh)
+        perm2 = await aget_user_permission_from_event(event2, gh)
 
         assert perm1 == Permission.WRITE
         assert perm2 == Permission.ADMIN
@@ -194,13 +218,42 @@ def test_sync_cache_hit(self):
         """Test that sync version uses cache."""
         gh = create_autospec(SyncGitHubAPI, instance=True)
         gh.getitem = Mock(return_value={"permission": "read"})
+        event = create_test_event("user", "owner", "repo")
 
         # First call should hit the API
-        perm1 = get_user_permission(gh, "owner", "repo", "user")
+        perm1 = get_user_permission_from_event(event, gh)
         assert perm1 == Permission.READ
         assert gh.getitem.call_count == 1
 
         # Second call should use cache
-        perm2 = get_user_permission(gh, "owner", "repo", "user")
+        perm2 = get_user_permission_from_event(event, gh)
         assert perm2 == Permission.READ
         assert gh.getitem.call_count == 1  # No additional API call
+
+
+class TestPermissionFromEvent:
+    @pytest.mark.asyncio
+    async def test_missing_comment_data(self):
+        """Test when event has no comment data."""
+        gh = create_autospec(AsyncGitHubAPI, instance=True)
+        event = sansio.Event({}, event="issue_comment", delivery_id="test")
+
+        permission = await aget_user_permission_from_event(event, gh)
+
+        assert permission == Permission.NONE
+        assert gh.getitem.called is False
+
+    @pytest.mark.asyncio
+    async def test_missing_repository_data(self):
+        """Test when event has no repository data."""
+        gh = create_autospec(AsyncGitHubAPI, instance=True)
+        event = sansio.Event(
+            {"comment": {"user": {"login": "user"}}},
+            event="issue_comment",
+            delivery_id="test",
+        )
+
+        permission = await aget_user_permission_from_event(event, gh)
+
+        assert permission == Permission.NONE
+        assert gh.getitem.called is False