Merge pull request #25 from desultory/clean_dict

Clean dict checks

Author: desultory

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ugrd"
-version = "1.14.3"
+version = "1.16.0"
 authors = [
   { name="Desultory", email="[email protected]" },
 ]
@@ -18,7 +18,7 @@ classifiers = [
 ]
 
 dependencies = [
-    "zenlib >= 2.1.2",
+    "zenlib >= 2.2.0",
     "pycpio >= 1.2.1"
 ]
 

src/ugrd/base/base.py

@@ -1,17 +1,16 @@
 __author__ = 'desultory'
-__version__ = '4.5.2'
+__version__ = '4.6.1'
 
 from importlib.metadata import version
 from pathlib import Path
 
-from zenlib.util import check_dict
+from zenlib.util import contains
 
 
-@check_dict('validate', value=True)
-@check_dict('hostonly', value=True)
+@contains('hostonly')
 def _validate_init_target(self) -> None:
     if not self['init_target'].exists():
-        raise FileNotFoundError('init_target not found at: %s', self['init_target'])
+        raise FileNotFoundError('init_target not found at: %s' % self['init_target'])
 
 
 def _process_init_target(self, target: Path) -> None:
@@ -22,18 +21,20 @@ def _process_init_target(self, target: Path) -> None:
     _validate_init_target(self)
 
 
-@check_dict('init_target', unset=True, message='init_target already set.')
+@contains('init_target', 'init_target is already set, skipping autodetection.', log_level=30)
 def _process_autodetect_init(self, state) -> None:
-    from shutil import which
     dict.__setitem__(self, 'autodetect_init', state)
-    if not state:
-        return
 
+
+@contains('autodetect_init', log_level=30)
+def autodetect_init(self) -> None:
+    """ Autodetects the init_target. """
+    from shutil import which
     if init := which('init'):
         self.logger.info('Detected init at: %s', init)
         self['init_target'] = init
     else:
-        raise FileNotFoundError('init_target is not specified and coud not be detected.')
+        raise FileNotFoundError('init_target is not specified and could not be detected.')
 
 
 def _find_init(self) -> str:
@@ -49,7 +50,7 @@ def _find_init(self) -> str:
             'return 1']
 
 
-@check_dict('init_target', not_empty=True, raise_exception=True, message='init_target must be set.')
+@contains('init_target', 'init_target must be set.', raise_exception=True)
 def do_switch_root(self) -> str:
     """
     Should be the final statement, switches root.

src/ugrd/base/base.toml

@@ -10,6 +10,9 @@ autodetect_init = true
 [imports.config_processing]
 "ugrd.base.base" = [ "_process_init_target", "_process_autodetect_init" ]
 
+[imports.build_pre]
+"ugrd.base.base" = [ "autodetect_init" ]
+
 [imports.init_final]
 "ugrd.base.base" = [ "do_switch_root" ]
 

src/ugrd/base/core.py

@@ -1,13 +1,13 @@
 __author__ = 'desultory'
-__version__ = '3.3.1'
+__version__ = '3.4.0'
 
 from pathlib import Path
 from typing import Union
 
-from zenlib.util import check_dict, NoDupFlatList
+from zenlib.util import contains, unset, NoDupFlatList
 
 
-@check_dict('clean', value=True, log_level=30, message="Skipping cleaning build directory")
+@contains('clean', "Skipping cleaning build directory", log_level=30)
 def clean_build_dir(self) -> None:
     """ Cleans the build directory. """
     from shutil import rmtree
@@ -119,7 +119,7 @@ def deploy_symlinks(self) -> None:
         self._symlink(symlink_parameters['source'], symlink_parameters['target'])
 
 
-@check_dict('mknod_cpio', value=False, log_level=20, message="Skipping real device node creation with mknod, as mknod_cpio is specified.")
+@unset('mknod_cpio', "Skipping real device node creation with mknod, as mknod_cpio is not specified.", log_level=20)
 def deploy_nodes(self) -> None:
     """ Generates specified device nodes. """
     from os import makedev, mknod
@@ -140,7 +140,7 @@ def deploy_nodes(self) -> None:
             raise e
 
 
-@check_dict('find_libgcc', value=True, log_level=20, message="Skipping libgcc_s dependency resolution.")
+@contains('find_libgcc', "Skipping libgcc_s dependency resolution", log_level=20)
 def find_libgcc(self) -> None:
     """
     Finds libgcc.so, adds a 'dependencies' item for it.

src/ugrd/base/debug.py

@@ -1,7 +1,7 @@
 __author__ = "desultory"
-__version__ = "1.2.1"
+__version__ = "1.3.0"
 
-from zenlib.util import check_dict
+from zenlib.util import contains
 
 
 def start_shell(self) -> str:
@@ -14,7 +14,7 @@ def start_shell(self) -> str:
             'bash -l']
 
 
-@check_dict('start_shell', value=True, message="Not enabling the debug shell, as the start_shell option is not set.")
+@contains('start_shell', 'Not enabling the debug shell, as the start_shell option is not set.', log_level=30)
 def enable_debug(self) -> str:
     """ Enable debug mode. """
     return "setvar debug 1"

src/ugrd/base/keymap.py

@@ -1,7 +1,7 @@
 __author__ = 'desultory'
-__version__ = '0.2.1'
+__version__ = '0.3.0'
 
-from zenlib.util import check_dict
+from zenlib.util import contains
 
 
 def _find_keymap_include(self, base_path, included_file, no_recurse=False):
@@ -71,7 +71,7 @@ def _process_keymap_file(self, keymap_file: str) -> str:
     dict.__setitem__(self, 'keymap_file', keymap_file.replace('.gz', ''))
 
 
-@check_dict('keymap_file', raise_exception=True, message="keymap_file must be set to use the keymap module")
+@contains('keymap_file', "keymap_file must be set to use the keymap module", raise_exception=True)
 def set_keymap(self) -> str:
     """ Sets the specified keymap. """
     return [f'einfo "Setting keymap: {self["keymap_file"]}"',

src/ugrd/crypto/cryptsetup.py

@@ -1,17 +1,19 @@
 __author__ = 'desultory'
-__version__ = '2.5.0'
+__version__ = '2.5.1'
 
-from zenlib.util import check_dict
+from zenlib.util import contains
 
 
 _module_name = 'ugrd.crypto.cryptsetup'
 
 CRYPTSETUP_PARAMETERS = ['key_type', 'partuuid', 'uuid', 'path', 'key_file', 'header_file', 'retries', 'key_command', 'reset_command', 'try_nokey', 'include_key']
 
 
-@check_dict('cryptsetup', value_arg=1, return_arg=2, contains=True)  # Check if the mapped name is defined
 def _merge_cryptsetup(self, mapped_name: str, config: dict) -> None:
     """ Merges the cryptsetup configuration """
+    if mapped_name not in self['cryptsetup']:
+        return config
+
     self.logger.log(5, "Existing cryptsetup configuration: %s" % self['cryptsetup'][mapped_name])
     self.logger.debug("[%s] Merging cryptsetup configuration: %s" % (mapped_name, config))
     return dict(self['cryptsetup'][mapped_name], **config)
@@ -38,7 +40,7 @@ def _process_cryptsetup_key_types_multi(self, key_type: str, config: dict) -> No
         self['cryptsetup_key_types'][key_type] = config
 
 
-@check_dict('validate', value=True, log_level=30, message="Skipping cryptsetup key validation.")
+@contains('validate', "Skipping cryptsetup keyfile validation.", log_level=30)
 def _validate_crypysetup_key(self, key_paramters: dict) -> None:
     """ Validates the cryptsetup key """
     if key_paramters.get('include_key'):
@@ -67,7 +69,7 @@ def _validate_crypysetup_key(self, key_paramters: dict) -> None:
         key_copy = parent
 
 
-@check_dict('validate', value=True, log_level=30, message="Skipping cryptsetup configuration validation.")
+@contains('validate', "Skipping cryptsetup configuration validation.", log_level=30)
 def _validate_cryptsetup_config(self, mapped_name: str, config: dict) -> None:
     self.logger.log(5, "[%s] Validating cryptsetup configuration: %s" % (mapped_name, config))
     for parameter in config:
@@ -115,7 +117,7 @@ def _process_cryptsetup_multi(self, mapped_name: str, config: dict) -> None:
     self['cryptsetup'][mapped_name] = config
 
 
-@check_dict('validate', value=True, log_level=30, message="Skipping LUKS source validation.")
+@contains('validate', "Skipping cryptsetup configuration validation.", log_level=30)
 def _validate_luks_source(self, mapped_name: str) -> None:
     """ Checks that a LUKS source device is valid """
     for _dm_info in self['_dm_info'].values():

src/ugrd/crypto/smartcard.py

@@ -1,18 +1,18 @@
 __author__ = 'desultory'
-__version__ = '1.0.0'
+__version__ = '1.1.0'
 
+from zenlib.util import contains
 
-from zenlib.util import check_dict
 
-
-@check_dict('sc_public_key', raise_exception=True, message="Smartcard public key file not specified (sc_public_key)")
+@contains('sc_public_key', "Smartcard public key file not specified (sc_public_key)", raise_exception=True)
 def fetch_keys(self) -> None:
+    """ Adds the GGP public key file to the list of dependencies. """
     self.logger.info("Adding GPG public key file to dependencies: %s", self['sc_public_key'])
     self['dependencies'] = self['sc_public_key']
 
 
-@check_dict('sc_public_key', raise_exception=True, message="Smartcard public key file not specified (sc_public_key)")
+@contains('sc_public_key', "Smartcard public key file not specified (sc_public_key)", raise_exception=True)
 def import_keys(self) -> str:
-    """ Import GPG public keys. """
+    """ Import GPG public keys at runtime. """
     return f"gpg --import {self['sc_public_key']}"
 

src/ugrd/fs/btrfs.py

@@ -1,8 +1,8 @@
-__version__ = '1.7.3'
+__version__ = '1.8.1'
 __author__ = 'desultory'
 
 
-from zenlib.util import check_dict
+from zenlib.util import contains, unset
 
 
 class SubvolNotFound(Exception):
@@ -25,16 +25,17 @@ def _get_mount_subvol(self, mountpoint: str) -> list:
     raise SubvolNotFound("No subvolume detected.")
 
 
-@check_dict('validate', value=True, message="Validate is not set, skipping root subvolume validation.")
-@check_dict('root_subvol', not_empty=True, message="root_subvol is not set, skipping validation.")
+@contains('validate', "validate is not enabled, skipping root subvolume validation.")
 def _validate_root_subvol(self) -> None:
     """ Validates the root subvolume. """
     try:
         detected_subvol = _get_mount_subvol(self, '/')
     except SubvolNotFound:
-        raise ValueError("Current root mount is not using a subvolume, but root_subvol is set: %s" % self['root_subvol'])
+        if self['root_subvol']:
+            raise ValueError("Current root mount is not using a subvolume, but root_subvol is set: %s" % self['root_subvol'])
     except SubvolIsRoot:
-        raise ValueError("Current root mount is not using a subvolume, but root_subvol is set: %s" % self['root_subvol'])
+        if self['root_subvol'] != '/':
+            raise ValueError("Current root mount is not using a subvolume, but root_subvol is set: %s" % self['root_subvol'])
 
     if self['root_subvol'] != detected_subvol:
         raise ValueError("[%s] Root subvolume does not match detected subvolume: %s" % (self['root_subvol'], detected_subvol))
@@ -62,10 +63,10 @@ def btrfs_scan(self) -> str:
     return 'einfo "$(btrfs device scan)"'
 
 
-@check_dict('subvol_selector', value=False, log_level=20, message="subvol_selector enabled, skipping.")
-@check_dict('autodetect_root_subvol', value=True, message="autodetect_root_subvol not enabled, skipping.")
-@check_dict('root_subvol', unset=True, message="root_subvol is set, skipping.")
-@check_dict('hostonly', value=True, message="hostonly is not set, skipping.")
+@unset('subvol_selector', message="subvol_selector is enabled, skipping.", log_level=20)
+@contains('autodetect_root_subvol', "autodetect_root_subvol is not enabled, skipping.", log_level=30)
+@unset('root_subvol', message="root_subvol is set, skipping.")
+@contains('hostonly', "hostonly is not enabled, skipping.", log_level=30)
 def autodetect_root_subvol(self):
     """ Detects the root subvolume. """
     try:
@@ -78,8 +79,8 @@ def autodetect_root_subvol(self):
         self.logger.debug("Root mount is not using a subvolume.")
 
 
-@check_dict('subvol_selector', value=True, message="subvol_selector not enabled, skipping")
-@check_dict('root_subvol', unset=True, message="root_subvol is set, skipping.")
+@contains('subvol_selector', message="subvol_selector is not enabled, skipping.")
+@unset('root_subvol', message="root_subvol is set, skipping.")
 def select_subvol(self) -> str:
     """ Returns a bash script to list subvolumes on the root volume. """
     # TODO: Figure out a way to make the case prompt more standard
@@ -106,7 +107,7 @@ def select_subvol(self) -> str:
             f"umount -l {self['_base_mount_path']}"]
 
 
-@check_dict('root_subvol', not_empty=True, message="root_subvol is not set, skipping.")
+@contains('root_subvol', message="root_subvol is not set, skipping.")
 def set_root_subvol(self) -> str:
     """ Adds the root_subvol to the root_mount options. """
     _validate_root_subvol(self)

src/ugrd/fs/lvm.py

@@ -1,7 +1,7 @@
 __author__ = 'desultory'
-__version__ = '1.2.0'
+__version__ = '1.2.1'
 
-from zenlib.util import check_dict
+from zenlib.util import contains
 
 
 def _process_lvm_multi(self, mapped_name: str, config: dict) -> None:
@@ -11,7 +11,7 @@ def _process_lvm_multi(self, mapped_name: str, config: dict) -> None:
     self['lvm'][mapped_name] = config
 
 
-@check_dict('lvm', not_empty=True, log_level=10, message="Skipping LVM initialization, no LVM configurations found.")
+@contains('lvm', "Skipping LVM initialization, no LVM configurations found.")
 def init_lvm(self) -> None:
     """ Returns bash lines to initialize LVM """
     return ['einfo "Initializing LVM, module version %s"' % __version__,