Refactor provisioners for better modularity

This refactors the provisioning logic so that every provisioner is
defined by its own class. This better encapsulates provisoner logic and
makes adding new provisioners easier.

While this is supposed to be a pure refactoring, it contains a small
functional change, so that provisioning scripts can now not just be
put as string only in the provisioner config, but following a "script"
key as well, similar to all other provisioners.

Author: Dunedan

incant/config_manager.py

@@ -1,5 +1,4 @@
 import os
-import re
 import sys
 from pathlib import Path
 from typing import Any, Dict, Optional
@@ -11,18 +10,22 @@
 from mako.template import Template
 
 from .exceptions import ConfigurationError
+from .incus_cli import IncusCLI
+from .provisioners import REGISTERED_PROVISIONERS
 from .reporter import Reporter
 from .types import InstanceConfig, InstanceDict
 
 
 class ConfigManager:
     def __init__(
         self,
+        incus: IncusCLI,
         reporter: Reporter,
         config_path: Optional[str] = None,
         verbose: bool = False,
         no_config: bool = False,
     ):
+        self.incus = incus
         self.reporter = reporter
         self.config_path = config_path
         self.verbose = verbose
@@ -130,13 +133,14 @@ def dump_config(self):
         except Exception as e:  # pylint: disable=broad-exception-caught
             raise ConfigurationError(f"Error dumping configuration: {e}") from e
 
-    def _validate_provision_step(self, step, step_idx, name):
+    def _validate_provision_step(self, step: Any, step_idx: int, name: str) -> None:
         if isinstance(step, str):
+            REGISTERED_PROVISIONERS.get("script")(self.incus, self.reporter).validate_config(name, step)
             return
 
         if not isinstance(step, dict):
             raise ConfigurationError(
-                f"Provisioning step {step_idx} in instance '{name}' " "must be a string or a dictionary."
+                f"Provisioning step {step_idx} in instance '{name}' must be a string or a dictionary."
             )
 
         if len(step) != 1:
@@ -147,102 +151,32 @@ def _validate_provision_step(self, step, step_idx, name):
 
         key, value = list(step.items())[0]
 
-        if key not in ["copy", "ssh", "llmnr"]:
+        if key not in REGISTERED_PROVISIONERS.keys():
             raise ConfigurationError(
                 f"Unknown provisioning step type '{key}' in instance '{name}'. "
-                "Accepted types are 'copy', 'ssh', or 'llmnr'."
+                f"Accepted types are {', '.join(REGISTERED_PROVISIONERS.keys())}."
             )
 
-        if key == "copy":
-            if not isinstance(value, dict):
-                raise ConfigurationError(
-                    f"Provisioning 'copy' step in instance '{name}' must have a dictionary value."
-                )
-            self._validate_copy_step(value, name)
-
-        if key == "ssh":
-            self._validate_ssh_step(value, name)
+        REGISTERED_PROVISIONERS.get(key)(self.incus, self.reporter).validate_config(name, value)
 
-        if key == "llmnr":
-            self._validate_llmnr_step(value, name)
-
-    def _validate_copy_step(self, value, name):
-        required_fields = ["source", "target"]
-        missing = [field for field in required_fields if field not in value]
-        if missing:
-            raise ConfigurationError(
-                (
-                    f"Provisioning 'copy' step in instance '{name}' is missing required "
-                    f"field(s): {', '.join(missing)}."
-                )
-            )
-        if not isinstance(value["source"], str) or not isinstance(value["target"], str):
-            raise ConfigurationError(
-                (f"Provisioning 'copy' step in instance '{name}' must have string " "'source' and 'target'.")
-            )
+    def _validate_provisioning(self, instance: InstanceConfig, name: str):
+        if instance.provision is None:
+            return
 
-        if "uid" in value and not isinstance(value["uid"], int):
-            raise ConfigurationError(
-                (f"Provisioning 'copy' step in instance '{name}' has invalid 'uid': " "must be an integer.")
-            )
-        if "gid" in value and not isinstance(value["gid"], int):
-            raise ConfigurationError(
-                (f"Provisioning 'copy' step in instance '{name}' has invalid 'gid': " "must be an integer.")
-            )
-        if "mode" in value:
-            mode_val = value["mode"]
-            if not isinstance(mode_val, str):
-                raise ConfigurationError(
-                    (
-                        f"Provisioning 'copy' step in instance '{name}' has invalid 'mode': "
-                        "must be a string like '0644'."
-                    )
-                )
-            if re.fullmatch(r"[0-7]{3,4}", mode_val) is None:
-                raise ConfigurationError(
-                    (
-                        f"Provisioning 'copy' step in instance '{name}' has invalid 'mode': "
-                        "must be 3-4 octal digits (e.g., '644' or '0644')."
-                    )
-                )
-        if "recursive" in value and not isinstance(value["recursive"], bool):
-            raise ConfigurationError(
-                (
-                    f"Provisioning 'copy' step in instance '{name}' has invalid 'recursive': "
-                    "must be a boolean."
-                )
-            )
-        if "create_dirs" in value and not isinstance(value["create_dirs"], bool):
-            raise ConfigurationError(
-                (
-                    f"Provisioning 'copy' step in instance '{name}' has invalid "
-                    "'create_dirs': must be a boolean."
-                )
-            )
+        provisions = instance.provision
 
-    def _validate_ssh_step(self, value, name):
-        if not isinstance(value, (bool, dict)):
-            raise ConfigurationError(
-                f"Provisioning 'ssh' step in instance '{name}' must have a boolean " "or dictionary value."
-            )
+        # Handle special "script" single-step provisioning.
+        if isinstance(provisions, str):
+            provisions = [provisions]
 
-    def _validate_llmnr_step(self, value, name):
-        if not isinstance(value, bool):
+        if isinstance(provisions, list):
+            for step_idx, step in enumerate(provisions):
+                self._validate_provision_step(step, step_idx, name)
+        else:
             raise ConfigurationError(
-                f"Provisioning 'llmnr' step in instance '{name}' must have a boolean value."
+                f"Provisioning for instance '{name}' must be a string or a list of steps."
             )
 
-    def _validate_provisioning(self, instance: InstanceConfig, name: str):
-        if instance.provision is not None:
-            provisions = instance.provision
-            if isinstance(provisions, list):
-                for step_idx, step in enumerate(provisions):
-                    self._validate_provision_step(step, step_idx, name)
-            elif not isinstance(provisions, str):
-                raise ConfigurationError(
-                    f"Provisioning for instance '{name}' must be a string or a list of steps."
-                )
-
     def _validate_pre_launch(self, instance: InstanceConfig, name: str):
         if instance.pre_launch_cmds is not None:
             pre_launch_cmds = instance.pre_launch_cmds
@@ -261,7 +195,6 @@ def validate_config(self):
             raise ConfigurationError("No instances found in config")
 
         for name, instance_config in self.instance_configs.items():
-
             # Validate 'provision' field
             self._validate_provisioning(instance_config, name)
             self._validate_pre_launch(instance_config, name)

incant/incant.py

@@ -15,18 +15,19 @@
 class Incant:
     def __init__(self, reporter: Reporter, **kwargs):
         self.reporter = reporter
+        self.incus = IncusCLI(self.reporter)
         self.verbose = kwargs.get("verbose", False)
         self.no_config = kwargs.get("no_config", False)
         self.config_manager = ConfigManager(
+            incus=self.incus,
             reporter=self.reporter,
             config_path=kwargs.get("config", None),
             verbose=self.verbose,
             no_config=self.no_config,
         )
         if not self.no_config:
             self.config_manager.validate_config()
-        self.incus = IncusCLI(self.reporter)
-        self.provisioner = ProvisionManager(self.incus, self.reporter)
+        self.provision_manager = ProvisionManager(self.incus, self.reporter)
 
     def _get_instance_configs(self, name: Optional[str] = None) -> InstanceDict:
         """Helper to get instance configs, either all or a specific one."""
@@ -91,7 +92,7 @@ def provision(self, name: Optional[str] = None):
 
         for instance_name, instance_config in instances_to_provision.items():
             if instance_config.provision:
-                self.provisioner.provision(instance_name, instance_config.provision)
+                self.provision_manager.provision(instance_name, instance_config.provision)
 
     def destroy(self, name=None):
         instances_to_destroy = self._get_instance_configs(name)
@@ -119,8 +120,7 @@ def list_instances(self, no_error: bool = False):
             self.reporter.echo(f"{instance_name}")
 
     def incant_init(self):
-        example_config = textwrap.dedent(
-            """
+        example_config = textwrap.dedent("""
           instances:
             basic-container:
               image: images:ubuntu/24.04
@@ -165,8 +165,7 @@ def incant_init(self):
                 provision:
                  - llmnr: true # LLMNR disabled by default on RHEL-based
                  - ssh: true
-            """
-        ).lstrip()
+            """).lstrip()
 
         config_path = "incant.yaml"
 

incant/provisioners/__init__.py

@@ -0,0 +1,7 @@
+from .base import REGISTERED_PROVISIONERS, Provisioner
+from .copy_file import CopyFile
+from .llmr import LLMR
+from .script import Script
+from .ssh_server import SSHServer
+
+__all__ = ["CopyFile", "LLMR", "Provisioner", "REGISTERED_PROVISIONERS", "Script", "SSHServer"]

incant/provisioners/base.py

@@ -0,0 +1,35 @@
+import abc
+import typing
+from typing import ClassVar, Union
+
+if typing.TYPE_CHECKING:
+    from incant import IncusCLI
+from incant.reporter import Reporter
+
+REGISTERED_PROVISIONERS: dict[str, type["Provisioner"]] = {}
+
+
+class Provisioner(abc.ABC):
+    """Abstract class, defining the interface for provisioners."""
+
+    # The name of the key used to identify this provisioner in the
+    # incant config.
+    config_key: ClassVar[str] = None
+
+    def __init__(self, incus_cli: "IncusCLI", reporter: Reporter):
+        self.incus = incus_cli
+        self.reporter = reporter
+
+    def __init_subclass__(cls, **kwargs):
+        super().__init_subclass__(**kwargs)
+        REGISTERED_PROVISIONERS[cls.config_key] = cls
+
+    @abc.abstractmethod
+    def validate_config(self, instance_name: str, config: Union[bool, dict, str]):
+        """Validate the given config."""
+        pass
+
+    @abc.abstractmethod
+    def provision(self, instance_name: str, config: Union[bool, dict, str]):
+        """Run the provisioning logic."""
+        pass

incant/provisioners/copy_file.py

@@ -0,0 +1,83 @@
+import re
+from typing import Union
+
+from ..exceptions import ConfigurationError
+from ..types import FilePushConfig
+from .base import Provisioner
+
+
+class CopyFile(Provisioner):
+    config_key = "copy"
+
+    def validate_config(self, instance_name: str, config: Union[bool, dict, str]):
+        if not isinstance(config, dict):
+            raise ConfigurationError(
+                f"Provisioning 'copy' step in instance '{instance_name}' must have a dictionary value."
+            )
+
+        required_fields = ["source", "target"]
+        missing = [field for field in required_fields if field not in config]
+        if missing:
+            raise ConfigurationError(
+                (
+                    f"Provisioning 'copy' step in instance '{instance_name}' is missing required "
+                    f"field(s): {', '.join(missing)}."
+                )
+            )
+        if not isinstance(config["source"], str) or not isinstance(config["target"], str):
+            raise ConfigurationError(
+                (
+                    f"Provisioning 'copy' step in instance '{instance_name}' "
+                    "must have string 'source' and 'target'."
+                )
+            )
+
+        if "uid" in config and not isinstance(config["uid"], int):
+            raise ConfigurationError(
+                (
+                    f"Provisioning 'copy' step in instance '{instance_name}' "
+                    "has invalid 'uid': must be an integer."
+                )
+            )
+        if "gid" in config and not isinstance(config["gid"], int):
+            raise ConfigurationError(
+                (
+                    f"Provisioning 'copy' step in instance '{instance_name}' "
+                    "has invalid 'gid': must be an integer."
+                )
+            )
+        if "mode" in config:
+            mode_val = config["mode"]
+            if not isinstance(mode_val, str):
+                raise ConfigurationError(
+                    (
+                        f"Provisioning 'copy' step in instance '{instance_name}' has invalid 'mode': "
+                        "must be a string like '0644'."
+                    )
+                )
+            if re.fullmatch(r"[0-7]{3,4}", mode_val) is None:
+                raise ConfigurationError(
+                    (
+                        f"Provisioning 'copy' step in instance '{instance_name}' has invalid 'mode': "
+                        "must be 3-4 octal digits (e.g., '644' or '0644')."
+                    )
+                )
+        if "recursive" in config and not isinstance(config["recursive"], bool):
+            raise ConfigurationError(
+                (
+                    f"Provisioning 'copy' step in instance '{instance_name}'"
+                    "has invalid 'recursive': must be a boolean."
+                )
+            )
+        if "create_dirs" in config and not isinstance(config["create_dirs"], bool):
+            raise ConfigurationError(
+                (
+                    f"Provisioning 'copy' step in instance '{instance_name}' "
+                    "has invalid 'create_dirs': must be a boolean."
+                )
+            )
+
+    def provision(self, instance_name: str, config: dict):
+        """Copy a file to the instance."""
+        config["instance_name"] = instance_name
+        self.incus.file_push(FilePushConfig(**config))