Merge pull request #814 from esune/feat/delete-credentials

Add explicit user confirmation to delete credential type

Author: WadeBarnes

server/vcr-server/agent_webhooks/management/commands/delete_credential_type.py

@@ -14,16 +14,22 @@ class Command(BaseCommand):
 
     def add_arguments(self, parser):
         parser.add_argument('credential_type_id', type=str)
+        parser.add_argument(
+            '--force',
+            action='store_true',
+            help='Skip confirmation prompt and proceed with deletion'
+        )
 
     def handle(self, *args, **options):
         credential_type_id = options['credential_type_id']
 
-        # Track affected topics for targeted reindexing
-        affected_topics = []
+        # Pre-flight check and confirmation before opening Solr queue
+        affected_topics = self._check_and_confirm_deletion(*args, **options)
 
+        # If we get here, user confirmed deletion - proceed with operations
         queue = SolrQueue()
         with queue:
-            affected_topics = self.delete_credential_type(*args, **options)
+            self._perform_deletion(affected_topics, *args, **options)
 
             # Wait for queue to drain while still in context
             eject = 20
@@ -45,10 +51,8 @@ def handle(self, *args, **options):
         self.stdout.write("Performing targeted search index refresh ...")
         self._refresh_affected_indexes(credential_type_id, affected_topics)
 
-    def delete_credential_type(self, *args, **options):
-        start_time = time.perf_counter()
-
-        # get Credential Type ID from input parameters
+    def _check_and_confirm_deletion(self, *args, **options):
+        """Check credential type exists and get user confirmation"""
         credential_type_id = options['credential_type_id']
         self.stdout.write("Deleting credential_type_id: " + credential_type_id)
 
@@ -60,7 +64,7 @@ def delete_credential_type(self, *args, **options):
             credential_type = CredentialType.objects.get(id=credential_type_id)
         except CredentialType.DoesNotExist:
             self.stdout.write(" ... credential_type_id not found in OrgBook.")
-            return list(affected_topics)
+            raise SystemExit(1)
 
         # Find all credentials for this credential type
         credentials = Credential.objects.filter(
@@ -71,10 +75,92 @@ def delete_credential_type(self, *args, **options):
         if credentials.exists():
             topic_ids = set(credentials.values_list('topic_id', flat=True))
             affected_topics.update(topic_ids)
+        
+        # Display information about the operation
+        self.stdout.write("\n" + "="*60)
+        self.stdout.write("CREDENTIAL TYPE DELETION CONFIRMATION")
+        self.stdout.write("="*60)
+        
+        # Use description if available, otherwise fall back to raw_data.type
+        display_name = credential_type.description
+        if (not display_name and hasattr(credential_type, 'raw_data') and
+                credential_type.raw_data):
+            try:
+                import json
+
+                # Parse JSON string to get the type attribute
+                raw_data_obj = json.loads(credential_type.raw_data)
+                display_name = raw_data_obj.get('type')
+            except (json.JSONDecodeError, AttributeError, TypeError, KeyError):
+                display_name = None
+        
+        # Final fallback to ID if no display name found
+        if not display_name:
+            display_name = f"Credential Type ID {credential_type_id}"
+            
+        self.stdout.write(f"Credential Type: {display_name}")
+        self.stdout.write(f"ID: {credential_type_id}")
+        self.stdout.write(
+            f"Found {credentials.count()} credentials to delete"
+        )
+        self.stdout.write(f"Topics affected: {len(affected_topics)}")
+        self.stdout.write("="*60)
+        
+        if affected_topics:
+            self.stdout.write(
+                f"WARNING: This operation will affect "
+                f"{len(affected_topics)} topics and trigger search index "
+                f"refresh."
+            )
+        else:
             self.stdout.write(
-                f" ... found {len(topic_ids)} topics affected by "
-                f"credential deletion"
+                "No topics will be affected by this operation."
             )
+            
+        self.stdout.write(
+            "\nThis action cannot be undone. All credentials of this type "
+            "will be permanently deleted."
+        )
+        
+        # Ask for user confirmation unless --force flag is used
+        if not options.get('force', False):
+            try:
+                confirmation = input(
+                    "\nDo you want to continue? (y/N): "
+                ).strip()
+            except (EOFError, KeyboardInterrupt):
+                self.stdout.write("\n\nOperation cancelled by user.")
+                raise SystemExit(0)
+            
+            if confirmation.lower() not in ['y', 'yes']:
+                self.stdout.write(
+                    "Operation cancelled. No changes were made."
+                )
+                raise SystemExit(0)
+        else:
+            self.stdout.write(
+                "\n--force flag provided, skipping confirmation."
+            )
+        
+        self.stdout.write("\nProceeding with credential type deletion...")
+        self.stdout.write("="*60 + "\n")
+        
+        return list(affected_topics)
+
+    def _perform_deletion(self, affected_topics, *args, **options):
+        """Perform the actual credential type deletion"""
+        start_time = time.perf_counter()
+        credential_type_id = options['credential_type_id']
+
+        # Re-fetch the credential type and related data (since confirmation
+        # happened earlier)
+        credential_type = CredentialType.objects.get(id=credential_type_id)
+        credentials = Credential.objects.filter(
+            credential_type=credential_type
+        )
+        
+        # Convert to set for efficient updates (we received it as a list)
+        affected_topics = set(affected_topics)
 
         # delete credentials from wallet first
         if credentials.exists():