Merge pull request #1261 from element-hq/gaelg/prompt-generic-function

Refactor to use a helper function for prompting

Author: gaelgatelement

packages/ess-migration-tool/newsfragments/1261.misc.md

@@ -0,0 +1 @@
+Refactor internal prompting logic.

packages/ess-migration-tool/src/ess_migration_tool/extra_files.py

@@ -15,7 +15,7 @@
 
 from .interfaces import ExtraFilesDiscoveryStrategy, SecretDiscoveryStrategy
 from .models import DiscoveredExtraFile, DiscoveredPath, SecretConfig
-from .utils import is_quiet_mode
+from .utils import is_quiet_mode, prompt_choice, prompt_value, prompt_yes_no
 
 logger = logging.getLogger("migration")
 
@@ -247,117 +247,111 @@ def prompt_for_missing_files(self) -> None:
             self.pretty_logger.info("  2. Skip these files and continue")
             self.pretty_logger.info("  3. Provide a directory to search for files")
 
-            while True:
-                try:
-                    choice = input("   Enter your choice (1/2/3): ").strip()
-
-                    if choice == "1":
-                        # Provide alternative paths
-                        self._prompt_for_individual_alternatives(file_path)
-                        break
-                    elif choice == "2":
-                        # Skip missing files
-                        self.pretty_logger.info("   ⚠️  Skipping missing files...")
-                        file_path.skipped_reason = "Skipped by user"
-                        break
-                    elif choice == "3":
-                        # Provide directory to search
-                        self._prompt_for_directory_search(file_path)
-                        break
-                    else:
-                        self.pretty_logger.info("   ❌ Invalid choice. Please enter 1, 2, or 3.")
-
-                except KeyboardInterrupt as err:
-                    self.pretty_logger.info("\n   ❌ Operation cancelled by user")
-                    raise ExtraFilesError("User cancelled file input") from err
-                except EOFError as err:
-                    self.pretty_logger.info("\n   ❌ End of input reached")
-                    raise ExtraFilesError("End of input reached during file prompt") from err
+            choice = prompt_choice(
+                self.pretty_logger,
+                "Enter your choice (1/2/3):",
+                [
+                    "Provide alternative path for this file",
+                    "Skip these files and continue",
+                    "Provide a directory to search for files",
+                ],
+            )
+
+            if choice == "Provide alternative path for this file":
+                # Provide alternative paths
+                self._prompt_for_individual_alternatives(file_path)
+            elif choice == "Skip these files and continue":
+                # Skip missing files
+                self.pretty_logger.info("   ⚠️  Skipping missing files...")
+                file_path.skipped_reason = "Skipped by user"
+            elif choice == "Provide a directory to search for files":
+                # Provide directory to search
+                self._prompt_for_directory_search(file_path)
 
     def _prompt_for_individual_alternatives(self, discovered_path: DiscoveredPath) -> None:
         """
         Prompt user for alternative paths for each missing file.
         """
-        while True:
+
+        def validate_file_path(path: str) -> tuple[bool, str]:
+            if path.lower() == "skip":
+                return True, ""  # Special case for skip
+            if not path:
+                return False, "File path cannot be empty. Please try again."
             try:
-                new_path = input("   Please enter the correct file path (or 'skip' to ignore): ").strip()
+                new_path_obj = Path(path)
+                if new_path_obj.is_dir():
+                    return False, "Targetted file is a directory."
+                discovered_extra_file = self._discover_extra_file(discovered_path, new_path_obj)
+                if not discovered_extra_file:
+                    return False, "File not found or invalid."
+                return True, ""
+            except ExtraFilesError as e:
+                return False, str(e)
 
-                if new_path.lower() == "skip":
-                    self.pretty_logger.info(f"   ⚠️  Skipping file: {discovered_path.source_path}")
-                    break
+        while True:
+            new_path = prompt_value(
+                self.pretty_logger,
+                "Please enter the correct file path (or 'skip' to ignore):",
+                validator=validate_file_path,
+            )
 
-                if not new_path:
-                    self.pretty_logger.info("   ❌ File path cannot be empty. Please try again.")
-                    continue
-
-                # Validate the new path
-                try:
-                    new_path_obj = Path(new_path)
-                    if new_path_obj.is_dir():
-                        raise ExtraFilesError("Targetted file is a directory.")
-                    else:
-                        discovered_extra_file = self._discover_extra_file(discovered_path, new_path_obj)
-                        if discovered_extra_file:
-                            self.discovered_extra_files[new_path_obj] = discovered_extra_file
-                            self.pretty_logger.info(f"   ✅ File validated: {new_path}")
-                            break
-                except ExtraFilesError as e:
-                    self.pretty_logger.info(f"   ❌ Invalid file path: {e}")
-                    self.pretty_logger.info("   Please try again or enter 'skip' to ignore.")
-
-            except KeyboardInterrupt as err:
-                self.pretty_logger.info("\n   ❌ Operation cancelled by user")
-                raise ExtraFilesError("User cancelled file input") from err
-            except EOFError as err:
-                self.pretty_logger.info("\n   ❌ End of input reached")
-                raise ExtraFilesError("End of input reached during file prompt") from err
+            if new_path.lower() == "skip":
+                self.pretty_logger.info(f"   ⚠️  Skipping file: {discovered_path.source_path}")
+                break
+
+            new_path_obj = Path(new_path)
+            discovered_extra_file = self._discover_extra_file(discovered_path, new_path_obj)
+            if discovered_extra_file:
+                self.discovered_extra_files[new_path_obj] = discovered_extra_file
+                self.pretty_logger.info(f"   ✅ File validated: {new_path}")
+                break
 
     def _prompt_for_directory_search(self, discovered_path: DiscoveredPath) -> None:
         """
         Prompt user for a directory to search for missing files.
         """
         self.pretty_logger.info("\n📁 Please provide a directory to search for missing files:")
 
-        while True:
+        def validate_directory(path: str) -> tuple[bool, str]:
+            if not path:
+                return False, "Directory path cannot be empty. Please try again."
             try:
-                search_dir = input("   Enter directory path: ").strip()
-
-                if not search_dir:
-                    self.pretty_logger.info("   ❌ Directory path cannot be empty. Please try again.")
-                    continue
-
-                # Validate the directory
-                try:
-                    dir_path = Path(search_dir)
-                    if not dir_path.is_dir():
-                        raise ExtraFilesError(f"Path is not a directory: {search_dir}")
-
-                    # Search for missing files in the directory
-                    found_files = self._handle_directory(discovered_path, dir_path)
-
-                    if found_files:
-                        self.pretty_logger.info(f"   ✅ Found {len(found_files)} matching files in {search_dir}")
-                        for file_path in found_files:
-                            self.pretty_logger.info(f"      📄 {Path(file_path).name}")
-                        break
-                    else:
-                        self.pretty_logger.info(f"   ⚠️  No matching files found in {search_dir}")
-                        self.pretty_logger.info("   Would you like to try another directory?")
-                        retry = input("   Enter 'yes' to try again or any other key to skip: ").strip().lower()
-                        if retry != "yes":
-                            self.pretty_logger.info("   ⚠️  Skipping directory search...")
-                            break
-
-                except Exception as e:
-                    self.pretty_logger.info(f"   ❌ Error searching directory: {e}")
-                    self.pretty_logger.info("   Please try again.")
-
-            except KeyboardInterrupt as err:
-                self.pretty_logger.info("\n   ❌ Operation cancelled by user")
-                raise ExtraFilesError("User cancelled file input") from err
-            except EOFError as err:
-                self.pretty_logger.info("\n   ❌ End of input reached")
-                raise ExtraFilesError("End of input reached during file prompt") from err
+                dir_path = Path(path)
+                if not dir_path.is_dir():
+                    return False, f"Path is not a directory: {path}"
+                return True, ""
+            except Exception as e:
+                return False, str(e)
+
+        while True:
+            search_dir = prompt_value(
+                self.pretty_logger,
+                "Enter directory path:",
+                validator=validate_directory,
+            )
+
+            dir_path = Path(search_dir)
+
+            # Search for missing files in the directory
+            found_files = self._handle_directory(discovered_path, dir_path)
+
+            if found_files:
+                self.pretty_logger.info(f"   ✅ Found {len(found_files)} matching files in {search_dir}")
+                for file_path in found_files:
+                    self.pretty_logger.info(f"      📄 {Path(file_path).name}")
+                break
+            else:
+                self.pretty_logger.info(f"   ⚠️  No matching files found in {search_dir}")
+                self.pretty_logger.info("   Would you like to try another directory?")
+                retry = prompt_yes_no(
+                    self.pretty_logger,
+                    "Try another directory?",
+                    default=False,
+                )
+                if not retry:
+                    self.pretty_logger.info("   ⚠️  Skipping directory search...")
+                    break
 
     def validate_extra_files(self) -> None:
         """

packages/ess-migration-tool/src/ess_migration_tool/secrets.py

@@ -11,7 +11,13 @@
 
 from .interfaces import SecretDiscoveryStrategy
 from .models import DiscoveredSecret, GlobalOptions
-from .utils import find_matching_schema_key, get_nested_value, is_quiet_mode, is_wildcard_pattern
+from .utils import (
+    find_matching_schema_key,
+    get_nested_value,
+    is_quiet_mode,
+    is_wildcard_pattern,
+    prompt_value,
+)
 
 logger = logging.getLogger("migration")
 
@@ -210,28 +216,19 @@ def prompt_for_missing_secrets(self) -> None:
             if not config_key:
                 raise RuntimeError(f"Missing configuration path for {secret_key}")
 
-            while True:
-                try:
-                    value = input("   Please paste the secret value: ").strip()
-                    if value:
-                        self.discovered_secrets[secret_key] = DiscoveredSecret(
-                            source_file=self.source_file,
-                            secret_key=secret_key,
-                            config_key=config_key,
-                            value=value,
-                        )
-                        # Remove from missing_required_secrets after successful prompt
-                        self.missing_required_secrets.remove((discovered_secret, error_message))
-                        self.pretty_logger.info(f"   ✅ Secret stored for {secret_key}")
-                        break
-                    else:
-                        self.pretty_logger.info("   ❌ Value cannot be empty. Please try again.")
-                except KeyboardInterrupt as err:
-                    self.pretty_logger.info("\n   ❌ Operation cancelled by user")
-                    raise SecretsError("User cancelled secret input") from err
-                except EOFError as err:
-                    self.pretty_logger.info("\n   ❌ End of input reached")
-                    raise SecretsError("End of input reached during secret prompt") from err
+            value = prompt_value(
+                self.pretty_logger,
+                "Please paste the secret value:",
+            )
+            self.discovered_secrets[secret_key] = DiscoveredSecret(
+                source_file=self.source_file,
+                secret_key=secret_key,
+                config_key=config_key,
+                value=value,
+            )
+            # Remove from missing_required_secrets after successful prompt
+            self.missing_required_secrets.remove((discovered_secret, error_message))
+            self.pretty_logger.info(f"   ✅ Secret stored for {secret_key}")
 
         self.pretty_logger.info(f"\n✅ All required {component_name} secrets have been provided")
         self.pretty_logger.info("=" * 60)

packages/ess-migration-tool/src/ess_migration_tool/synapse.py

@@ -16,7 +16,7 @@
 from .interfaces import ExtraFilesDiscoveryStrategy, SecretDiscoveryStrategy
 from .migration import ConfigValueTransformer, MigrationStrategy, TransformationSpec, additional_config_transformer
 from .models import DiscoveredSecret, GlobalOptions, MigrationError, SecretConfig
-from .utils import extract_hostname_from_url, yaml_dump_with_pipe_for_multiline
+from .utils import extract_hostname_from_url, prompt_choice, prompt_value, yaml_dump_with_pipe_for_multiline
 
 logger = logging.getLogger("migration")
 
@@ -67,27 +67,12 @@ def prompt_user_for_worker(
     for i, worker_type in enumerate(matched_worker_types):
         config_value_transformer.pretty_logger.info(f"   ❌   {i + 1}. {worker_type}")
 
-    while True:
-        try:
-            value = input(f"   Please select the worker type of instance {instance_name}: ").strip()
-            if value:
-                # make sure user selected an valid integer
-                try:
-                    worker_index = int(value) - 1
-                    if worker_index in range(len(matched_worker_types)):
-                        return matched_worker_types[worker_index]
-                    else:
-                        config_value_transformer.pretty_logger.info(f"Invalid worker type: {value}")
-                except ValueError:
-                    config_value_transformer.pretty_logger.info("   ❌ Please select a valid worker type.")
-            else:
-                config_value_transformer.pretty_logger.info("   ❌ Value cannot be empty. Please try again.")
-        except KeyboardInterrupt as err:
-            config_value_transformer.pretty_logger.info("\n   ❌ Operation cancelled by user")
-            raise MigrationError("User cancelled worker input") from err
-        except EOFError as err:
-            config_value_transformer.pretty_logger.info("\n   ❌ End of input reached")
-            raise MigrationError("End of input reached during worker prompt") from err
+    selected_worker = prompt_choice(
+        config_value_transformer.pretty_logger,
+        f"Please select the worker type of instance {instance_name}:",
+        matched_worker_types,
+    )
+    return selected_worker
 
 
 def extract_workers_from_instance_map(
@@ -163,19 +148,10 @@ def prompt_for_ingress_host(
     )
     config_value_transformer.pretty_logger.info("   ❌ Please provide Synapse ingress host (e.g., matrix.example.com):")
 
-    while True:
-        try:
-            ingress_host = input("   Enter ingress host: ").strip()
-            if ingress_host:
-                return ingress_host
-            else:
-                config_value_transformer.pretty_logger.info("   ❌ Ingress host cannot be empty. Please try again.")
-        except KeyboardInterrupt as err:
-            config_value_transformer.pretty_logger.info("\n   ❌ Operation cancelled by user")
-            raise MigrationError("User cancelled ingress host input") from err
-        except EOFError as err:
-            config_value_transformer.pretty_logger.info("\n   ❌ End of input reached")
-            raise MigrationError("End of input reached during ingress host prompt") from err
+    return prompt_value(
+        config_value_transformer.pretty_logger,
+        "Enter ingress host:",
+    )
 
 
 def filter_listeners(_, listeners: list[dict] | None, **kwargs: Any) -> dict[str, Any] | None: