Bringing more granularity to reset command option

src/cli/onefuzz/api.py

@@ -1290,70 +1290,164 @@ def config(
 
         return data
 
-    def reset(self, everything: bool = False) -> None:
-        """
-        Resets onefuzz.  Stops all jobs, notifications, and repro VMs.
-        Specifying 'everything' will delete all containers, pools, and managed
-        scalesets.
-        """
-        message = ["Confirm stopping *all* running jobs"]
-        if everything:
-            message += ["AND deleting *ALL* fuzzing related data"]
-        message += ["(specify y or n): "]
-
+    def _user_confirmation(self, message: str) -> bool:
         answer: Optional[str] = None
         while answer not in ["y", "n"]:
             answer = input(" ".join(message)).strip()
 
         if answer == "n":
             self.logger.info("not stopping")
-            return
-
-        self.logger.info("stopping all interactions")
+            return False
+        return True
 
+    def _delete_jobs(self):
         for job in self.jobs.list():
             self.logger.info("stopping job %s", job.job_id)
             self.jobs.delete(job.job_id)
 
+    def _delete_tasks(self):
         for task in self.tasks.list():
             self.logger.info("stopping task %s", task.task_id)
             self.tasks.delete(task.task_id)
 
+    def _delete_notifications(self):
         for notification in self.notifications.list():
             self.logger.info("stopping notification %s", notification.notification_id)
             self.notifications.delete(notification.notification_id)
 
+    def _delete_repros(self):
         for vm in self.repro.list():
             self.repro.delete(str(vm.vm_id))
 
+    def _delete_pools(self):
+        for pool in self.pools.list():
+            self.logger.info("stopping pool: %s", pool.name)
+            self.pools.shutdown(pool.name, now=True)
+
+    def _delete_scalesets(self):
+        for scaleset in self.scalesets.list():
+            self.logger.info("stopping scaleset: %s", scaleset.scaleset_id)
+            self.scalesets.shutdown(scaleset.scaleset_id, now=True)
+
+    def _delete_containers(self):
+        for container in self.containers.list():
+            if (
+                container.metadata
+                and "container_type" in container.metadata
+                and container.metadata["container_type"]
+                in [
+                    "analysis",
+                    "coverage",
+                    "crashes",
+                    "inputs",
+                    "no_repro",
+                    "readonly_inputs",
+                    "reports",
+                    "setup",
+                    "unique_reports",
+                ]
+            ):
+                self.logger.info("removing container: %s", container.name)
+                self.containers.delete(container.name)
+
+    def _delete_all_running_jobs(self):
+        self._delete_jobs()
+        self._delete_tasks()
+        self._delete_notifications()
+        self._delete_repros()
+
+    def _delete_fuzzing_data(self):
+        self._delete_scalesets()
+        self._delete_pools()
+        self._delete_containers()
+
+    def reset(
+        self,
+        *,
+        containers: bool = False,
+        everything: bool = False,
+        jobs: bool = False,
+        notifications: bool = False,
+        pools: bool = False,
+        repros: bool = False,
+        scalesets: bool = False,
+        tasks: bool = False,
+        yes: bool = False,
+    ) -> None:
+        """
+        Resets onefuzz. Stops all jobs, notifications, and repro VMs.
+        Specifying 'everything' will delete all containers, pools, and managed
+        scalesets.
+
+        :param bool containers: Delete all the containers.
+        :param bool everything: Delete all containers, pools and managed scalesets.
+        :param bool jobs: Delete all jobs.
+        :param bool notifications: Delete all notifications.
+        :param bool pools: Delete all pools.
+        :param bool repros: Delete all repro vms.
+        :param bool scalesets: Delete all managed scalesets.
+        :param bool tasks: Delete all tasks.
+        :param bool yes: Ignoring to specify "y" in prompt.
+        """
+        message = ["Confirm stopping *all* running jobs"]
         if everything:
-            for pool in self.pools.list():
-                self.logger.info("stopping pool: %s", pool.name)
-                self.pools.shutdown(pool.name, now=True)
-
-            for scaleset in self.scalesets.list():
-                self.logger.info("stopping scaleset: %s", scaleset.scaleset_id)
-                self.scalesets.shutdown(scaleset.scaleset_id, now=True)
-
-            for container in self.containers.list():
-                if (
-                    container.metadata
-                    and "container_type" in container.metadata
-                    and container.metadata["container_type"]
-                    in [
-                        "analysis",
-                        "coverage",
-                        "crashes",
-                        "inputs",
-                        "no_repro",
-                        "readonly_inputs",
-                        "reports",
-                        "setup",
-                        "unique_reports",
-                    ]
-                ):
-                    self.logger.info("removing container: %s", container.name)
-                    self.containers.delete(container.name)
+            message += ["AND deleting *ALL* fuzzing related data"]
+        message += ["(specify y or n): "]
+
+        delete_options = [
+            jobs,
+            tasks,
+            notifications,
+            pools,
+            scalesets,
+            repros,
+            containers,
+        ]
+        arguments = [everything]
+        arguments.extend(delete_options)
+        if not any(arguments):
+            if not yes and not self._user_confirmation(message):
+                return
+            self._delete_all_running_jobs()
+            return
+
+        if everything:
+            if not yes and not self._user_confirmation(message):
+                return
+            self._delete_all_running_jobs()
+            self._delete_fuzzing_data()
+            return
+
+        to_delete = []
+        for k, _ in locals().items():
+            if k in delete_options:
+                to_delete.append(k)
+        print(to_delete)
+        message = ["Confirm stopping %s running jobs" % (", ".join(to_delete))]
+        message += ["(specify y or n): "]
+        if not yes and not self._user_confirmation(message):
+            return
+
+        if jobs:
+            self._delete_jobs()
+
+        if tasks:
+            self._delete_tasks()
+
+        if notifications:
+            self._delete_notifications()
+
+        if repros:
+            self._delete_repros()
+
+        if pools:
+            self._delete_pools()
+
+        if scalesets:
+            self._delete_scalesets()
+
+        if containers:
+            self._delete_containers()
 
 
 from .debug import Debug  # noqa: E402