split into functions

src/gradio_space_ci/webhook.py

@@ -256,21 +256,9 @@ async def trigger_ci_on_pr(payload: WebhookPayload, task_queue: BackgroundTasks)
         and payload.discussion.isPullRequest
         and payload.discussion.status == "open"
     ):
-        # A comment, is it by a trusted author ?
+        # A comment, is it by a command ?
         if payload.event.scope == "discussion.comment":
-            pr_num = payload.discussion.num
-            details = get_discussion_details(repo_id=space_id, repo_type="space", discussion_num=pr_num)
-            event_author = details.events[-1]._event["author"]["name"]  # username of that event
-            if event_author in EPHEMERAL_SPACES_CONFIG["trusted_authors"]:
-                print("Comment detected with content:\n", payload.comment.content)
-                if payload.comment.content == "/trust_pr":
-                    print("trusting pr ...")
-                    set_config(space_id=space_id, pr_num=pr_num)
-                    print("pr has been trusted")
-                elif payload.comment.content == "/untrust_pr":
-                    print("untrusting pr ...")
-                    unset_config(space_id=space_id, pr_num=pr_num)
-                    print("pr has been untrusted")
+            handle_command(space_id=space_id, payload=payload)
         # Always sync (in case the space was sleeping or building)
         if not is_pr_synced(space_id=space_id, pr_num=payload.discussion.num):
             # New PR! Sync task scheduled
@@ -298,12 +286,8 @@ async def trigger_ci_on_pr(payload: WebhookPayload, task_queue: BackgroundTasks)
         # => loop through all PRs and check if new changes happened
         for discussion in get_repo_discussions(repo_id=space_id, repo_type="space"):
             if discussion.is_pull_request and discussion.status == "open":
-                # Was the pr updated by an untrusted author ?
-                details = get_discussion_details(repo_id=space_id, repo_type="space", discussion_num=discussion.num)
-                event_author = details.events[-1]._event["author"]["name"]  # username of that event
-                if event_author not in EPHEMERAL_SPACES_CONFIG["trusted_authors"]:
-                    # Untrusted author, we unset the config as part or security reasons
-                    unset_config(space_id=space_id, pr_num=discussion.num)
+                # If untrusted author made a modification we unset the config
+                handle_modification(space_id=space_id, discussion=discussion)
                 if not is_pr_synced(space_id=space_id, pr_num=discussion.num):
                     # Found a PR that is not yet synced
                     task_queue.add_task(sync_ci_space, space_id=space_id, pr_num=discussion.num)
@@ -509,6 +493,33 @@ def unset_config(space_id: str, pr_num: int) -> None:
     delete_space_storage(ci_space_id)
 
 
+def handle_modification(space_id: str, discussion: Any) -> None:
+    ci_space_id = _get_ci_space_id(space_id=space_id, pr_num=discussion.num)
+    if not repo_exists(ci_space_id):
+        return
+    details = get_discussion_details(repo_id=space_id, repo_type="space", discussion_num=discussion.num)
+    event_author = details.events[-1]._event["author"]["name"]  # username of that event
+    if event_author not in EPHEMERAL_SPACES_CONFIG["trusted_authors"]:
+        # Untrusted author, we unset the config as part or security reasons
+        unset_config(space_id=space_id, pr_num=discussion.num)
+
+
+def handle_command(space_id: str, payload: WebhookPayload) -> None:
+    """when a trusted author writes a command we handle it"""
+    pr_num = payload.discussion.num
+    details = get_discussion_details(repo_id=space_id, repo_type="space", discussion_num=pr_num)
+    event_author = details.events[-1]._event["author"]["name"]  # username of that event
+    if event_author in EPHEMERAL_SPACES_CONFIG["trusted_authors"]:
+        if payload.comment.content == "/trust_pr":
+            print("trusting pr ...")
+            set_config(space_id=space_id, pr_num=pr_num)
+            print("pr has been trusted")
+        elif payload.comment.content == "/untrust_pr":
+            print("untrusting pr ...")
+            unset_config(space_id=space_id, pr_num=pr_num)
+            print("pr has been untrusted")
+
+
 NOTIFICATION_TEMPLATE_CREATED_AND_CONFIGURED = """\
 Following the creation of this PR, an ephemeral Space [{ci_space_id}](https://huggingface.co/spaces/{ci_space_id}) has been started. Any changes pushed to this PR will be synced with the test Space.
 Since this PR has been created by a trusted author, the ephemeral Space has been configured with the correct hardware, storage, and secrets.