Allow users to init the config (or chunk) from a custom location

With this change, users can now specify a `--from` argument with a valid
path or url to a valid broker settings file.
This will still try to fallback on the current behavior of using a local
repo or the main SatelliteQE/broker repo as the source.

broker/commands.py

@@ -475,12 +475,15 @@ def restore():
 
 @loggedcli(group=config)
 @click.argument("chunk", type=str, required=False)
-def init(chunk):
+@click.option("--from", "_from", type=str, help="A file path or URL to initialize the config from.")
+def init(chunk=None, _from=None):
     """Initialize the broker configuration file from your local clone or GitHub.
 
     You can also init specific chunks by passing the chunk name.
+    Additionally, if you want to initialize from a file or URL, you can pass the `--from` flag.
+    Keep in mind that the file and url contents need to be valid yaml.
     """
-    ConfigManager(settings.settings_path).init_config_file(chunk)
+    ConfigManager(settings.settings_path).init_config_file(chunk=chunk, _from=_from)
 
 
 @loggedcli(group=config)
@@ -503,9 +506,10 @@ def nick(nick, no_syntax):
 
 
 @loggedcli(group=config)
-def migrate():
+@click.option("-f", "--force-version", type=str, help="Force the migration to a specific version")
+def migrate(force_version=None):
     """Migrate the broker configuration file to the latest version."""
-    ConfigManager(settings.settings_path).migrate()
+    ConfigManager(settings.settings_path).migrate(force_version=force_version)
 
 
 @loggedcli(group=config)

broker/config_manager.py

@@ -110,16 +110,22 @@ def _import_config(self, source, is_url=False):
         else:
             return source.read_text()
 
-    def _get_migrations(self):
+    def _get_migrations(self, force_version=None):
         """Construct a list of all applicable migrations."""
         from broker import config_migrations
 
         config_version = Version(self._cfg.get("_version", "0.0.0"))
+        if force_version:
+            force_version = Version(force_version)
         migrations = []
         for _, name, _ in pkgutil.iter_modules(config_migrations.__path__):
             module = importlib.import_module(f"broker.config_migrations.{name}")
-            if hasattr(module, "run_migrations") and config_version < file_name_to_ver(name):
-                migrations.append(module)
+            if hasattr(module, "run_migrations"):
+                if force_version and force_version == file_name_to_ver(name):
+                    migrations.append(module)
+                    break
+                elif config_version < file_name_to_ver(name):
+                    migrations.append(module)
         return migrations
 
     def backup(self):
@@ -199,26 +205,34 @@ def nicks(self, nick=None):
             return nicks[nick]
         return list(nicks.keys())
 
-    def init_config_file(self, chunk=None):
+    def init_config_file(self, chunk=None, _from=None):
         """Check for the existence of the config file and create it if it doesn't exist."""
         if self.interactive_mode and self._settings_path.exists() and not chunk:
             # if the file exists, ask the user if they want to overwrite it
             if (
                 click.prompt(
-                    f"Settings file already exists at {self._settings_path.absolute()}. Overwrite?",
+                    f"Overwrite the settings file at {self._settings_path.absolute()}. Overwrite?",
                     type=click.Choice(["y", "n"]),
                     default="n",
                 )
                 != "y"
             ):
                 return
-        # get the example file from the local repo or GitHub
-        example_path = Path(__file__).parent.parent.joinpath("broker_settings.yaml.example")
         raw_data = None
-        if example_path.exists():
-            raw_data = self._import_config(example_path)
+        if _from:
+            # determine if this is a local file or a URL
+            if Path(_from).exists():
+                raw_data = self._import_config(Path(_from))
+            else:
+                raw_data = self._import_config(_from, is_url=True)
+        # if we still don't have data, get the example file from the local repo or GitHub
         if not raw_data:
-            raw_data = self._import_config(GH_CFG, is_url=True)
+            # get the example file from the local repo or GitHub
+            example_path = Path(__file__).parent.parent.joinpath("broker_settings.yaml.example")
+            if example_path.exists():
+                raw_data = self._import_config(example_path)
+            if not raw_data:
+                raw_data = self._import_config(GH_CFG, is_url=True)
         if not raw_data:
             raise exceptions.ConfigurationError(
                 f"Broker settings file not found at {self._settings_path.absolute()}."
@@ -228,10 +242,10 @@ def init_config_file(self, chunk=None):
             chunk_data = self._interactive_edit(chunk_data)
         self.update(chunk, chunk_data)
 
-    def migrate(self):
+    def migrate(self, force_version=None):
         """Migrate the config from a previous version of Broker."""
         # get all available migrations
-        if not (migrations := self._get_migrations()):
+        if not (migrations := self._get_migrations(force_version)):
             logger.info("No migrations are applicable to your config.")
             return
         # run all migrations in order