basic commands

src/gradio_space_ci/webhook.py

@@ -28,6 +28,9 @@ from huggingface_hub import (
     snapshot_download,
     space_info,
     upload_folder,
+    delete_space_secret,
+    delete_space_variable,
+    delete_space_storage
 )
 from huggingface_hub.repocard import RepoCard
 from huggingface_hub.utils import (
@@ -248,16 +251,7 @@ async def trigger_ci_on_pr(payload: WebhookPayload, task_queue: BackgroundTasks)
 
     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 (
-        # Means comment by a trusted author
-        payload.event.scope == "discussion.comment"
-        and payload.event.action == "create"
-        and payload.discussion.isPullRequest
-        and payload.discussion.status == "open"
-        and event_author in EPHEMERAL_SPACES_CONFIG["trusted_authors"]
-    ):
-        print("Comment detected with content:\n ", payload.comment.content)
+    event_author = details.events[-1]._event["author"]["name"]  # username of that event
 
     if (
         # Means "a new PR has been opened"
@@ -295,8 +289,28 @@ async def trigger_ci_on_pr(payload: WebhookPayload, task_queue: BackgroundTasks)
             if discussion.is_pull_request and discussion.status == "open":
                 if not is_pr_synced(space_id=space_id, pr_num=discussion.num):
                     # Found a PR that is not yet synced
+                    if event_author not in EPHEMERAL_SPACES_CONFIG["trusted_authors"] : 
+                        # if change by untrusted author unset config
+                        unset_config(space_id=space_id,pr_num=pr_num)
                     task_queue.add_task(sync_ci_space, space_id=space_id, pr_num=discussion.num)
                     has_task = True
+    if (
+        # Means comment by a trusted author
+        payload.event.scope == "discussion.comment"
+        and payload.event.action == "create"
+        and payload.discussion.isPullRequest
+        and payload.discussion.status == "open"
+        and 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")
 
     if has_task:
         return Response("Task scheduled to sync/delete Space", status_code=status.HTTP_202_ACCEPTED)
@@ -463,6 +477,42 @@ def _get_ci_space_id(space_id: str, pr_num: int) -> str:
     return f"{space_id}-ci-pr-{pr_num}"
 
 
+def set_config(space_id: str, pr_num: str) -> None:
+    """a function to set the ephemerial space config"""
+    variables: Dict[str, str] = EPHEMERAL_SPACES_CONFIG["variables"]
+    secrets: Dict[str, str] = EPHEMERAL_SPACES_CONFIG["secrets"]
+    hardware: Optional[SpaceHardware] = EPHEMERAL_SPACES_CONFIG["hardware"]
+    storage: Optional[SpaceHardware] = EPHEMERAL_SPACES_CONFIG["storage"]
+    ci_space_id = _get_ci_space_id(space_id=space_id, pr_num=pr_num)
+    # Configure space
+    for key, value in variables.items():
+        add_space_variable(ci_space_id, key, value)
+    for key, value in secrets.items():
+        add_space_secret(ci_space_id, key, value)
+
+    # Request hardware/storage for space
+    if hardware is not None and hardware != SpaceHardware.CPU_BASIC:
+        request_space_hardware(ci_space_id, hardware, sleep_time=5 * 60)  # sleep after 5min on PR Spaces with GPU
+    if storage is not None:
+        request_space_storage(ci_space_id, storage)
+
+
+def unset_config(space_id: str, pr_num: int) -> None:
+    "a function to unset the configuration of an ephemeral space"
+    ci_space_id = _get_ci_space_id(space_id=space_id, pr_num=pr_num)
+    variables: Dict[str, str] = EPHEMERAL_SPACES_CONFIG["variables"]
+    secrets: Dict[str, str] = EPHEMERAL_SPACES_CONFIG["secrets"]
+    # Unset space variables and secrets
+    for key  in variables.keys():
+        delete_space_variable(ci_space_id, key)
+    for key in secrets.keys(): 
+        delete_space_secret(ci_space_id, key)
+    # Reset hardware
+    request_space_hardware(ci_space_id,SpaceHardware.CPU_BASIC)
+    delete_space_storage(ci_space_id)
+
+
+
 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.