Wip/initrd separation

cleanroom/binarymanager.py

@@ -17,33 +17,36 @@
 class Binaries(Enum):
     """Important binaries."""
 
+    APT_GET = auto()
     BORG = auto()
     BTRFS = auto()
-    MKNOD = auto()
-    PACMAN = auto()
-    PACMAN_KEY = auto()
-    APT_GET = auto()
-    DPKG = auto()
+    CHROOT_HELPER = auto()
+    CPIO = auto()
     DEBOOTSTRAP = auto()
-    SBSIGN = auto()
-    OBJCOPY = auto()
-    MKSQUASHFS = auto()
-    VERITYSETUP = auto()
-    TAR = auto()
-    USERMOD = auto()
-    USERADD = auto()
+    DEPMOD = auto()
+    DPKG = auto()
+    FIND = auto()
+    FLOCK = auto()
     GROUPADD = auto()
     GROUPMOD = auto()
-    CHROOT_HELPER = auto()
-    SYSTEMCTL = auto()
-    SFDISK = auto()
-    FLOCK = auto()
+    MKFS_VFAT = auto()
+    MKNOD = auto()
+    MKSQUASHFS = auto()
+    MODPROBE = auto()
+    NBD_CLIENT = auto()
+    OBJCOPY = auto()
+    PACMAN = auto()
+    PACMAN_KEY = auto()
     QEMU_IMG = auto()
     QEMU_NBD = auto()
-    NBD_CLIENT = auto()
-    MKFS_VFAT = auto()
+    SBSIGN = auto()
+    SFDISK = auto()
     SYNC = auto()
-    MODPROBE = auto()
+    SYSTEMCTL = auto()
+    TAR = auto()
+    USERADD = auto()
+    USERMOD = auto()
+    VERITYSETUP = auto()
 
 
 def _check_for_binary(binary: str) -> str:
@@ -67,40 +70,43 @@ def _find_binaries() -> typing.Dict[Binaries, str]:
     binaries = {
         Binaries.BORG: _check_for_binary("/usr/bin/borg"),
         Binaries.BTRFS: _check_for_binary("/usr/bin/btrfs"),
-        Binaries.SBSIGN: _check_for_binary("/usr/bin/sbsign"),
-        Binaries.OBJCOPY: _check_for_binary("/usr/bin/objcopy"),
-        Binaries.MKNOD: _check_for_binary("/usr/bin/mknod"),
-        Binaries.MKSQUASHFS: _check_for_binary("/usr/bin/mksquashfs"),
-        Binaries.TAR: _check_for_binary("/usr/bin/tar"),
-        Binaries.USERMOD: _check_for_binary("/usr/sbin/usermod"),
-        Binaries.USERADD: _check_for_binary("/usr/sbin/useradd"),
-        Binaries.GROUPMOD: _check_for_binary("/usr/sbin/groupmod"),
-        Binaries.GROUPADD: _check_for_binary("/usr/sbin/groupadd"),
         Binaries.CHROOT_HELPER: _check_for_binary("/usr/bin/arch-chroot"),
-        Binaries.SYSTEMCTL: _check_for_binary("/usr/bin/systemctl"),
+        Binaries.CPIO: _check_for_binary("/usr/bin/cpio"),
+        Binaries.DEPMOD: _check_for_binary("/usr/bin/depmod"),
+        Binaries.FIND: _check_for_binary("/usr/bin/find"),
         Binaries.FLOCK: _check_for_binary("/usr/bin/flock"),
-        Binaries.SFDISK: _check_for_binary("/usr/bin/sfdisk"),
+        Binaries.GROUPADD: _check_for_binary("/usr/sbin/groupadd"),
+        Binaries.GROUPMOD: _check_for_binary("/usr/sbin/groupmod"),
+        Binaries.MKFS_VFAT: _check_for_binary("/usr/bin/mkfs.vfat"),
+        Binaries.MKNOD: _check_for_binary("/usr/bin/mknod"),
+        Binaries.MKSQUASHFS: _check_for_binary("/usr/bin/mksquashfs"),
+        Binaries.MODPROBE: _check_for_binary("/usr/bin/modprobe"),
+        Binaries.NBD_CLIENT: _check_for_binary("/usr/bin/nbd-client"),
+        Binaries.OBJCOPY: _check_for_binary("/usr/bin/objcopy"),
         Binaries.QEMU_IMG: _check_for_binary("/usr/bin/qemu-img"),
         Binaries.QEMU_NBD: _check_for_binary("/usr/bin/qemu-nbd"),
-        Binaries.NBD_CLIENT: _check_for_binary("/usr/bin/nbd-client"),
-        Binaries.MKFS_VFAT: _check_for_binary("/usr/bin/mkfs.vfat"),
+        Binaries.SBSIGN: _check_for_binary("/usr/bin/sbsign"),
+        Binaries.SFDISK: _check_for_binary("/usr/bin/sfdisk"),
         Binaries.SYNC: _check_for_binary("/usr/bin/sync"),
-        Binaries.MODPROBE: _check_for_binary("/usr/bin/modprobe"),
+        Binaries.SYSTEMCTL: _check_for_binary("/usr/bin/systemctl"),
+        Binaries.TAR: _check_for_binary("/usr/bin/tar"),
+        Binaries.USERADD: _check_for_binary("/usr/sbin/useradd"),
+        Binaries.USERMOD: _check_for_binary("/usr/sbin/usermod"),
     }
     os_binaries: typing.Dict[Binaries, str] = {}
     distribution = _get_distribution()
     debug("Distribution: {}".format(distribution))
     if distribution == "debian":
         os_binaries = {
             Binaries.APT_GET: _check_for_binary("/usr/bin/apt-get"),
-            Binaries.DPKG: _check_for_binary("/usr/bin/dpkg"),
             Binaries.DEBOOTSTRAP: _check_for_binary("/usr/sbin/debootstrap"),
+            Binaries.DPKG: _check_for_binary("/usr/bin/dpkg"),
             Binaries.VERITYSETUP: _check_for_binary("/usr/sbin/veritysetup"),
         }
     elif distribution == "arch" or distribution == "archlinux":
         os_binaries = {
-            Binaries.PACMAN: _check_for_binary("/usr/bin/pacman"),
             Binaries.PACMAN_KEY: _check_for_binary("/usr/bin/pacman-key"),
+            Binaries.PACMAN: _check_for_binary("/usr/bin/pacman"),
             Binaries.VERITYSETUP: _check_for_binary("/usr/bin/veritysetup"),
         }
     else:
@@ -116,15 +122,19 @@ class BinaryManager:
     def __init__(self) -> None:
         """Constructor."""
         self._binaries = _find_binaries()
+        self._optionals: typing.List[Binaries] = []
 
     def preflight_check(self) -> None:
         passed = True
         for b in self._binaries.items():
             if b[1]:
                 debug("{} found: {}...".format(b[0], b[1]))
             else:
-                warn("{} not found.".format(b[0]))
-                passed = False
+                if b in self._optionals:
+                    debug("[OPTIONAL] {} not found, ignoring.")
+                else:
+                    warn("{} not found.".format(b[0]))
+                    passed = False
         if not passed:
             raise PreflightError("Required binaries are not available.")
 

cleanroom/command.py

@@ -14,7 +14,7 @@
 from .exceptions import GenerateError, ParseError
 from .execobject import ExecObject
 from .location import Location
-from .printer import fail, h3, success, verbose
+from .printer import debug, fail, h3, success, verbose
 from .systemcontext import SystemContext
 
 import os
@@ -47,18 +47,23 @@ def __init__(
         file: str,
         syntax: str = "",
         help_string: str,
-        **services: typing.Any
+        **services: typing.Any,
     ) -> None:
         """Constructor."""
         self._name = name
         self._syntax_string = syntax
         self._help_string = help_string
         helper_directory = os.path.join(
-            os.path.dirname(os.path.realpath(file)), "helper", self._name
+            os.path.dirname(os.path.realpath(file)), os.path.basename(file)[:-3],
         )
         self.__helper_directory = (
             helper_directory if os.path.isdir(helper_directory) else None
         )
+        if self.__helper_directory is None:
+            debug(f"Checked {helper_directory} for helpers for command {name}: NONE")
+        else:
+            debug(f"Checked {helper_directory} for helpers for command {name}: FOUND")
+
         self._services = services
 
     @property
@@ -103,7 +108,7 @@ def __call__(
         location: Location,
         system_context: SystemContext,
         *args: typing.Any,
-        **kwargs: typing.Any
+        **kwargs: typing.Any,
     ) -> None:
         """Implement this!
 
@@ -117,7 +122,7 @@ def _execute(
         system_context: SystemContext,
         command: str,
         *args: typing.Any,
-        **kwargs: typing.Any
+        **kwargs: typing.Any,
     ) -> None:
         command_info = self._service("command_manager").command(command)
         if not command_info:
@@ -132,7 +137,7 @@ def _add_hook(
         hook_name: str,
         command: str,
         *args: typing.Any,
-        **kwargs: typing.Any
+        **kwargs: typing.Any,
     ) -> None:
         """Add a hook."""
         command_info = self._service("command_manager").command(command)
@@ -170,7 +175,7 @@ def _helper_directory(self) -> typing.Optional[str]:
         """Return the helper directory."""
         return self.__helper_directory
 
-    def _config_directory(self, system_context) -> str:
+    def _config_directory(self, system_context: SystemContext) -> str:
         return os.path.join(
             system_context.systems_definition_directory, "config", self.name
         )
@@ -188,7 +193,7 @@ def _validate_arguments_exact(
         arg_count: int,
         message: str,
         *args: typing.Any,
-        **kwargs: typing.Any
+        **kwargs: typing.Any,
     ) -> None:
         self._validate_args_exact(location, arg_count, message, *args)
         self._validate_kwargs(location, (), **kwargs)
@@ -199,7 +204,7 @@ def _validate_arguments_at_least(
         arg_count: int,
         message: str,
         *args: typing.Any,
-        **kwargs: typing.Any
+        **kwargs: typing.Any,
     ) -> None:
         self._validate_args_at_least(location, arg_count, message, *args)
         self._validate_kwargs(location, (), **kwargs)
@@ -223,7 +228,7 @@ def _validate_kwargs(
         self,
         location: Location,
         known_kwargs: typing.Tuple[str, ...],
-        **kwargs: typing.Any
+        **kwargs: typing.Any,
     ) -> None:
         if not known_kwargs:
             if kwargs:
@@ -244,7 +249,7 @@ def _require_kwargs(
         self,
         location: Location,
         required_kwargs: typing.Tuple[str, ...],
-        **kwargs: typing.Any
+        **kwargs: typing.Any,
     ) -> None:
         for key in required_kwargs:
             if key not in kwargs:

cleanroom/commandmanager.py

@@ -8,48 +8,58 @@
 from .command import Command, stringify
 from .exceptions import PreflightError
 from .location import Location
-from .printer import debug, error, h2, success, trace
-from .systemcontext import SystemContext, _recursive_expand
+from .printer import debug, h2, success, trace
+from .systemcontext import SystemContext
 
-import collections
 import importlib.util
 import inspect
 import os
 import re
 import typing
 
 
-CommandInfo = collections.namedtuple(
-    "CommandInfo",
-    [
-        "name",
-        "syntax_string",
-        "help_string",
-        "file_name",
-        "dependency_func",
-        "validate_func",
-        "execute_func",
-        "register_substitutions",
-    ],
-)
+class CommandInfo(typing.NamedTuple):
+    name: str
+    syntax_string: str
+    help_string: str
+    file_name: str
+    dependency_func: typing.Callable[
+        [typing.Tuple[typing.Any, ...], typing.Dict[str, typing.Any]],
+        typing.Optional[str],
+    ]
+    validate_func: typing.Callable[
+        [Location, typing.Tuple[typing.Any, ...], typing.Dict[str, typing.Any],], None,
+    ]
+    execute_func: typing.Callable[
+        [
+            Location,
+            SystemContext,
+            typing.Tuple[typing.Any, ...],
+            typing.Dict[str, typing.Any],
+        ],
+        None,
+    ]
+    register_substitutions: typing.Callable[
+        [], typing.List[typing.Tuple[str, str, str]]
+    ]
 
 
 def _process_args(system_context: SystemContext, *args: typing.Any) -> typing.Any:
-    return tuple(map(lambda a: _recursive_expand(system_context, a), args))
+    return tuple(map(lambda a: system_context.expand(a), args))
 
 
 def _process_kwargs(
     system_context: SystemContext, **kwargs: typing.Any
 ) -> typing.Dict[str, typing.Any]:
-    return {k: _recursive_expand(system_context, v) for k, v in kwargs.items()}
+    return {k: system_context.expand(v) for k, v in kwargs.items()}
 
 
 def call_command(
     location: Location,
     system_context: SystemContext,
     command: Command,
     *args: typing.Any,
-    **kwargs: typing.Dict[str, typing.Any]
+    **kwargs: typing.Any
 ):
     _args = _process_args(system_context, *args)
     _kwargs = _process_kwargs(system_context, **kwargs)
@@ -82,20 +92,33 @@ def print_commands(self) -> None:
                 )
             )
 
-    def _collect_substitutions(self) -> typing.List[typing.Tuple[str, str, str, str]]:
-        result: typing.List[typing.Tuple[str, str, str, str]] = []
-        duplications: typing.Dict[str, str] = {}
+    def _collect_substitutions(
+        self,
+    ) -> typing.List[typing.Tuple[str, str, str, typing.Tuple[str, ...]]]:
+        result: typing.Dict[
+            str, typing.Tuple[str, str, str, typing.Tuple[str, ...]]
+        ] = {}
+
         for cmd in self._commands.keys():
             command_info = self.command(cmd)
             assert command_info
 
             name = command_info.name
             for (key, value, description) in command_info.register_substitutions():
-                result.append((key, value, description, name))
-                assert not key in duplications
-                duplications[key] = name
-
-        return result
+                if not key in result:
+                    result[key] = (key, value, description, (name,))
+                else:
+                    (old_key, old_value, old_description, old_names) = result[key]
+                    assert (
+                        old_key == key
+                        and old_value == value
+                        and old_description == description
+                        and not name in old_names
+                        and old_names
+                    )
+                    result[key] = (key, value, description, (*old_names, name))
+
+        return [v for v in result.values()]
 
     def print_substitutions(self) -> None:
         h2("Predefined Substitutions:")
@@ -117,10 +140,10 @@ def print_substitutions(self) -> None:
             print('  {} ("{}"): {}\n    {}\n'.format(key, value, name, description))
 
     def setup_substitutions(self, system_context: SystemContext):
-        if system_context._base_context:
+        if system_context.base_context:
             debug(
                 'System Context inherited, using substitutions from "{}".'.format(
-                    system_context._base_context.system_name
+                    system_context.base_context.system_name
                 )
             )
             return
@@ -216,7 +239,7 @@ def _find_commands_in_directory(self, directory: str) -> None:
             assert spec and spec.loader
             spec.loader.exec_module(cmd_module)
 
-            def is_command(x):
+            def is_command(x: typing.Any) -> bool:
                 return (
                     inspect.isclass(x)
                     and x.__name__.endswith("Command")