mypy fixes and always import fs image

LICENSE.md

@@ -99,7 +99,7 @@ extent that it includes a convenient and prominently visible feature that **(1)*
 displays an appropriate copyright notice, and **(2)** tells the user that there is no
 warranty for the work (except to the extent that warranties are provided), that
 licensees may convey the work under this License, and how to view a copy of this
-License. If the interface presents a list of user commands or options, such as a
+License. If the interface presents a list of user print_commands or options, such as a
 menu, a prominent item in the list meets this criterion.
 
 ### 1. Source Code
@@ -124,7 +124,7 @@ the executable work runs, or a compiler used to produce the work, or an object c
 interpreter used to run it.
 
 The “Corresponding Source” for a work in object code form means all the
-source code needed to generate, install, and (for an executable work) run the object
+source code needed to generate_systems, install, and (for an executable work) run the object
 code and to modify the work, including scripts to control those activities. However,
 it does not include the work's System Libraries, or general-purpose tools or
 generally available free programs which are used unmodified in performing those
@@ -578,8 +578,8 @@ when it starts in an interactive mode:
     This is free software, and you are welcome to redistribute it
     under certain conditions; type 'show c' for details.
 
-The hypothetical commands `show w` and `show c` should show the appropriate parts of
-the General Public License. Of course, your program's commands might be different;
+The hypothetical print_commands `show w` and `show c` should show the appropriate parts of
+the General Public License. Of course, your program's print_commands might be different;
 for a GUI interface, you would use an “about box”.
 
 You should also get your employer (if you work as a programmer) or school, if any, to

cleanroom/generator/preflight.py → cleanroom/binarymanager.py

@@ -1,15 +1,39 @@
 # -*- coding: utf-8 -*-
-"""The Context the generation will run in.
+"""An object used to find/get host system binaries.
 
 @author: Tobias Hunger <[email protected]>
 """
 
 
-from .context import Binaries, Context
-from ..exceptions import PreflightError
-from ..printer import (debug, error, info, h2, warn,)
+from .printer import fail, info, trace, warn
 
+from enum import Enum, auto, unique
 import os
+import typing
+
+
+@unique
+class Binaries(Enum):
+    """Important binaries."""
+
+    BORG = auto()
+    BTRFS = auto()
+    PACMAN = auto()
+    PACMAN_KEY = auto()
+    PACSTRAP = auto()
+    APT_GET = auto()
+    DPKG = auto()
+    DEBOOTSTRAP = auto()
+    SBSIGN = auto()
+    OBJCOPY = auto()
+    MKSQUASHFS = auto()
+    VERITYSETUP = auto()
+    TAR = auto()
+    USERMOD = auto()
+    USERADD = auto()
+    GROUPADD = auto()
+    GROUPMOD = auto()
+    CHROOT_HELPER = auto()
 
 
 def _check_for_binary(binary: str) -> str:
@@ -18,15 +42,15 @@ def _check_for_binary(binary: str) -> str:
 
 
 def _get_distribution():
-    with open("/usr/lib/os-release") as os:
-        for line in os.readlines():
+    with open("/usr/lib/os-release") as os_release:
+        for line in os_release.readlines():
             line = line.strip()
             if line.startswith('ID_LIKE='):
                 return line[8:].strip('"')
     return "<UNSUPPORTED>"
 
 
-def _find_binaries(ctx: Context) -> None:
+def _find_binaries() -> typing.Dict[Binaries, str]:
     binaries = {
         Binaries.BORG: _check_for_binary('/usr/bin/borg'),
         Binaries.BTRFS: _check_for_binary('/usr/bin/btrfs'),
@@ -41,57 +65,50 @@ def _find_binaries(ctx: Context) -> None:
         # in arch-install-scripts in ubuntu:-)
         Binaries.CHROOT_HELPER: _check_for_binary('/usr/bin/arch-chroot'),
     }
-    os_binaries = {}
+    os_binaries: typing.Dict[Binaries, str] = {}
     distribution = _get_distribution()
-    if (distribution == "debian"):
+    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.VERITYSETUP: _check_for_binary('/usr/sbin/veritysetup'),
         }
-    elif (distribution == "archlinux"):
+    elif distribution == "archlinux":
         os_binaries = {
             Binaries.PACMAN: _check_for_binary('/usr/bin/pacman'),
             Binaries.PACMAN_KEY: _check_for_binary('/usr/bin/pacman-key'),
             Binaries.PACSTRAP: _check_for_binary('/usr/bin/pacstrap'),
             Binaries.VERITYSETUP: _check_for_binary('/usr/bin/veritysetup'),
         }
     else:
-        error("Unsupported Linux flavor.")
+        fail("Unsupported Linux flavor.")
 
-    ctx.set_binaries({**binaries, **os_binaries})
+    return {**binaries, **os_binaries}
 
 
-def preflight_check(ctx: Context) -> None:
-    """Run a fast pre-flight check on the context."""
-    h2('Running Preflight Checks.', verbosity=2)
+class BinaryManager:
+    """The find and allow access to all the different system binaries
+       Cleanroom may need."""
 
-    _find_binaries(ctx)
+    def __init__(self) -> None:
+        """Constructor."""
+        self._binaries = _find_binaries()
 
-    binaries = _preflight_binaries_check(ctx)
-    users = _preflight_users_check(ctx)
+    def preflight_check(self) -> bool:
+        passed = True
+        for b in self._binaries.items():
+            if b[1]:
+                info('{} found: {}...'.format(b[0], b[1]))
+            else:
+                warn('{} not found.'.format(b[0]))
+                passed = False
+        return passed
 
-    if not binaries or not users:
-        raise PreflightError('Preflight Check failed.')
+    def binary(self, selector: Binaries) -> typing.Optional[str]:
+        """Get a binary from the context."""
+        assert self._binaries
 
-
-def _preflight_binaries_check(ctx: Context) -> bool:
-    """Check executables."""
-    passed = True
-    for b in ctx._binaries.items():
-        if b[1]:
-            info('{} found: {}...'.format(b[0], b[1]))
-        else:
-            warn('{} not found.'.format(b[0]))
-            passed = False
-    return passed
-
-
-def _preflight_users_check(ctx: Context) -> bool:
-    """Check tha the script is running as root."""
-    if os.geteuid() == 0:
-        debug('Running as root.')
-        return True
-    warn('Not running as root.')
-    return False
+        binary = self._binaries[selector]
+        trace('Binary for {}: {}.'.format(selector, binary))
+        return binary

cleanroom/command.py

@@ -0,0 +1,176 @@
+# -*- coding: utf-8 -*-
+"""Base class for print_commands usable in the system definition files.
+
+The Command class will be used to derive all system definition file print_commands
+from.
+
+@author: Tobias Hunger <[email protected]>
+"""
+
+
+from __future__ import annotations
+
+from .exceptions import GenerateError, ParseError
+from .location import Location
+from .printer import trace, fail, success
+from .systemcontext import SystemContext
+
+import os
+import os.path
+import typing
+
+
+ServicesType = typing.Mapping[str, typing.Any]
+
+
+def _stringify(command: str, *args, **kwargs):
+    args_str = ' "' + '" "'.join(args) + '"' if args else ''
+    kwargs_str = ' '.join(map(lambda kv: kv[0] + '="' + str(kv[1]) + '"',
+                              kwargs.items())) if kwargs else ''
+    return '"{}"'.format(command) + args_str + kwargs_str
+
+
+class Command:
+    """A command that can be used in to set up a machines."""
+
+    def __init__(self, name: str, *, file: str,
+                 syntax: str = '', help_string: str,
+                 services: ServicesType) \
+            -> 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)
+        self.__helper_directory = helper_directory if os.path.isdir(helper_directory) else None
+        self._services = services
+
+    @property
+    def syntax_string(self) -> str:
+        """Return syntax description."""
+        if self._syntax_string:
+            return '{} {}'.format(self._name, self._syntax_string)
+        return self._name
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def help_string(self) -> str:
+        """Print help string."""
+        return self._help_string
+
+    def validate(self, location: Location,
+                 *args: typing.Any, **kwargs: typing.Any) -> None:
+        """Implement this!
+
+        Validate all arguments.
+        """
+        fail('Command "{}"" called validate_arguments illegally!'
+             .format(self.name))
+        return None
+
+    def dependency(self, *args: typing.Any, **kwargs: typing.Any) \
+            -> typing.Optional[str]:
+        """Maybe implement this, but this default implementation should be ok."""
+        return None
+
+    def execute(self, location: Location, system_context: SystemContext,
+                *args: typing.Any, **kwargs: typing.Any) -> None:
+        command_str = _stringify(self.name, *args, **kwargs)
+        trace('{}: Execute {}.'.format(location, self.name, command_str))
+        self(location, system_context, *args, **kwargs)
+        success('{}: Executed {}{}'.format(location, self.name, command_str))
+
+    def __call__(self, location: Location, system_context: SystemContext,
+                 *args: typing.Any, **kwargs: typing.Any) -> None:
+        """Implement this!"""
+
+    def _execute(self, location: Location, system_context: SystemContext,
+                 command: str, *args: typing.Any, **kwargs: typing.Any) -> None:
+        command_str = _stringify(command, *args, **kwargs)
+        trace('{}: Execute {}.'.format(location, self.name, command_str))
+
+        command_info = self.service('command_manager').command(command)
+        if not command_info:
+            raise GenerateError('Command "{}" not found.'.format(command))
+        command_info.validate_func(location, *args, **kwargs)
+        command_info.execute_func(location, system_context, *args, **kwargs)
+        success('{}: Executed {}{}'.format(location, self.name, command_str))
+
+    def service(self, service_name: str) -> typing.Any:
+        return self._services.get(service_name, None)
+
+    @property
+    def _helper_directory(self) -> typing.Optional[str]:
+        """Return the helper directory."""
+        return self.__helper_directory
+
+    def _validate_no_arguments(self, location: Location,
+                               *args: typing.Any, **kwargs: typing.Any) -> None:
+        self._validate_no_args(location, *args)
+        self._validate_kwargs(location, (), **kwargs)
+
+    def _validate_arguments_exact(self, location: Location, arg_count: int, message: str,
+                                  *args: typing.Any, **kwargs: typing.Any) -> None:
+        self._validate_args_exact(location, arg_count, message, *args)
+        self._validate_kwargs(location, (), **kwargs)
+
+    def _validate_arguments_at_least(self, location: Location, arg_count: int, message: str,
+                                     *args: typing.Any, **kwargs: typing.Any) -> None:
+        self._validate_args_at_least(location, arg_count, message, *args)
+        self._validate_kwargs(location, (), **kwargs)
+
+    def _validate_no_args(self, location: Location, *args: typing.Any) -> None:
+        if args is list:
+            trace('Validating arguments: "{}".'.format('", "'.join(str(args))))
+        else:
+            trace('Validating argument: "{}".'.format(args))
+        self._validate_args_exact(location, 0,
+                                  '"{}" does not take arguments.', *args)
+
+    def _validate_args_exact(self, location: Location, arg_count: int,
+                             message: str, *args: typing.Any) -> None:
+        if args is list:
+            trace('Validating arguments: "{}".'.format('", "'.join(str(args))))
+        else:
+            trace('Validating argument: "{}".'.format(args))
+        if len(args) != arg_count:
+            raise ParseError(message.format(self.name), location=location)
+
+    def _validate_args_at_least(self, location: Location, arg_count: int,
+                                message: str, *args: typing.Any) -> None:
+        if args is list:
+            trace('Validating arguments: "{}".'.format('", "'.join(str(args))))
+        else:
+            trace('Validating argument: "{}".'.format(args))
+        if len(args) < arg_count:
+            raise ParseError(message.format(self.name), location=location)
+
+    def _validate_kwargs(self, location: Location, known_kwargs: typing.Tuple[str, ...],
+                         **kwargs: typing.Any) -> None:
+        trace('Validating keyword arguments: "{}"'
+              .format('", "'.join(['{}={}'.format(k, str(kwargs[k]))
+                                   for k in kwargs.keys()])))
+        if not known_kwargs:
+            if kwargs:
+                raise ParseError('"{}" does not accept keyword arguments.'
+                                 .format(self.name), location=location)
+        else:
+            for key, value in kwargs.items():
+                if key not in known_kwargs:
+                    raise ParseError('"{}" does not accept the keyword '
+                                     'arguments "{}".'
+                                     .format(self.name, key),
+                                     location=location)
+
+    def _require_kwargs(self, location: Location, required_kwargs: typing.Tuple[str, ...],
+                        **kwargs: typing.Any) -> None:
+        for key in required_kwargs:
+            if key not in kwargs:
+                raise ParseError('"{}" requires the keyword '
+                                 'arguments "{}" to be passed.'
+                                 .format(self.name, key),
+                                 location=location)