Merge pull request #22 from desultory/tests

Add basic tests support

Author: desultory

src/ugrd/base/cmdline.py

@@ -1,5 +1,5 @@
 __author__ = 'desultory'
-__version__ = '2.2.3'
+__version__ = '2.3.0'
 
 
 CMDLINE_BOOLS = ['quiet', 'debug', 'recovery', 'rootwait']
@@ -60,4 +60,9 @@ def mount_cmdline_root(self) -> str:
 
 def export_exports(self) -> list:
     """ Returns a bash script exporting all exports defined in the exports key. """
+    from importlib.metadata import version, PackageNotFoundError
+    try:
+        self['exports']['VERSION'] = version(__package__)
+    except PackageNotFoundError:
+        self['exports']['VERSION'] = 9999
     return [f'setvar {key} "{value}"' for key, value in self['exports'].items()]

src/ugrd/base/core.toml

@@ -10,6 +10,7 @@ library_paths = [ "/lib64" ]
 old_count = 1
 
 binaries = [ "/bin/bash" ]
+banner = 'einfo "UGRD v$(readvar VERSION)"'
 
 [nodes.console]
 mode = 0o644
@@ -64,4 +65,5 @@ out_dir = "Path"  # The directory where the initramfs is packed/output. If no pa
 old_count = "int"  # The number of times to cycle old files before deleting
 clean = "bool"  # Add the clean property, used to define if the mounts should be cleaned up after boot
 file_owner = "str"  # Add the file_owner property, used to define who should own the copied initramfs files
+banner = "str"  # Banner string to be printed at the start of init
 

src/ugrd/crypto/cryptsetup.py

@@ -67,6 +67,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.")
 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:

src/ugrd/fs/cpio.py

@@ -1,5 +1,5 @@
 __author__ = 'desultory'
-__version__ = '2.9.0'
+__version__ = '3.0.1'
 
 
 from pycpio import PyCPIO
@@ -14,8 +14,8 @@ def check_cpio(self, cpio: PyCPIO) -> None:
             self.logger.debug("Dependency found in CPIO: %s" % dep)
 
 
-def get_cpio_filename(self) -> str:
-    """ Generates a CPIO filename based on the current configuration. """
+def get_archive_path(self) -> str:
+    """ Determines the filename for the output CPIO archive based on the current configuration. """
     if out_file := self.get('out_file'):
         self.logger.info("Using specified out_file: %s" % out_file)
     else:
@@ -28,15 +28,17 @@ def get_cpio_filename(self) -> str:
             if compression_type.lower() != 'false':  # The variable is a string, so we need to check for the string 'false'
                 out_file += f".{compression_type}"
 
-    return self.out_dir / out_file
+    self['_archive_out_path'] = self.out_dir / out_file
 
 
 def make_cpio(self) -> None:
     """
-    Creates a CPIO archive from the build directory and writes it to the output directory.
+    Populates the CPIO archive using the build directory,
+    writes it to the output file, and rotates the output file if necessary.
+    Creates device nodes in the CPIO archive if the mknod_cpio option is set.
     Raises FileNotFoundError if the output directory does not exist.
     """
-    cpio = PyCPIO(logger=self.logger, _log_bump=10)
+    cpio = self._cpio_archive
     cpio.append_recursive(self.build_dir, relative=True)
     check_cpio(self, cpio)
 
@@ -45,7 +47,7 @@ def make_cpio(self) -> None:
             self.logger.debug("Adding CPIO node: %s" % node)
             cpio.add_chardev(name=node['path'], mode=node['mode'], major=node['major'], minor=node['minor'])
 
-    out_cpio = get_cpio_filename(self)
+    out_cpio = self['_archive_out_path']
 
     if not out_cpio.parent.exists():
         self._mkdir(out_cpio.parent)
@@ -80,5 +82,7 @@ def _process_out_file(self, out_file):
         self['out_dir'] = Path(out_file).parent
         self.logger.info("Resolved out_dir to: %s" % self['out_dir'])
         out_file = Path(out_file).name
+    else:
+        out_file = Path(out_file)
 
     dict.__setitem__(self, 'out_file', out_file)

src/ugrd/fs/cpio.toml

@@ -5,6 +5,9 @@ cpio_rotate = true
 [imports.config_processing]
 "ugrd.fs.cpio" = [ "_process_out_file" ]
 
+[imports.build_tasks]
+"ugrd.fs.cpio" = ['get_archive_path']
+
 [imports.pack]
 "ugrd.fs.cpio" = [ "make_cpio" ]
 
@@ -13,3 +16,5 @@ out_file = "str"  # The name of the cpio file to create.
 cpio_rotate = "bool"  # makes a .old backup of the cpio file if it already exists.
 mknod_cpio = "bool"  # When enabled, mknod is not used to create device nodes, they are just created in the cpio.
 cpio_compression = "str"  # The compression method to use for the cpio file. Currently, only xz is supported.
+_archive_out_path = "Path"  # The name of the file to create, determined based on runtime config.
+_cpio_archive = "PyCPIO"  # The cpio archive object.

src/ugrd/fs/test_image.py

@@ -0,0 +1,31 @@
+__version__ = "0.3.0"
+
+
+def init_test_vars(self):
+    if not self.get('test_kernel'):
+        raise ValueError("No test kernel specified")
+    elif not self['test_kernel'].exists():
+        raise FileNotFoundError("Test kernel not found: %s" % self['test_kernel'])
+
+    self['banner'] = f"echo {self['test_flag']}"
+
+
+def make_test_image(self):
+    """ Creates a test image from the build dir """
+    self.logger.info("Creating test image from: %s" % self.build_dir.resolve())
+
+    # Create the test image file, flll with 0s
+    with open(self._archive_out_path, "wb") as f:
+        self.logger.info("Creating test image file: %s" % self._archive_out_path)
+        f.write(b"\0" * self.test_image_size * 2 ** 20)
+
+    rootfs_uuid = self['mounts']['root']['uuid']
+    rootfs_type = self['mounts']['root']['type']
+
+    if rootfs_type == 'ext4':
+        self._run(['mkfs', '-t', rootfs_type, '-d', self.build_dir.resolve(), '-U', rootfs_uuid, '-F', self._archive_out_path])
+    elif rootfs_type == 'btrfs':
+        self._run(['mkfs', '-t', rootfs_type, '--rootdir', self.build_dir.resolve(), '-U', rootfs_uuid, self._archive_out_path])
+    else:
+        raise Exception("Unsupported test rootfs type: %s" % rootfs_type)
+

src/ugrd/fs/test_image.toml

@@ -0,0 +1,36 @@
+modules = ['ugrd.fs.cpio']
+paths = ['dev', 'sys', 'proc']
+
+
+binaries = ['halt', 'shutdown', 'ls', 'top', 'ps']
+shebang = "#!/bin/bash"
+
+qemu_arch = 'x86_64'
+test_image_size = 16
+test_memory = '256M'
+test_cpu = 'host'
+test_cmdline = 'console=ttyS0,115200 panic=1'
+test_flag = "UGRD TEST SUCCESSFUL"
+test_timeout = 15
+
+[masks]
+pack = "make_cpio"
+
+[imports.build_pre]
+"ugrd.fs.test_image" = ["init_test_vars"]
+
+[imports.pack]
+"ugrd.fs.test_image" = ["make_test_image"]
+
+[custom_parameters]
+shebang = "str"  # Add the shebang property, because the test mode disables the base module
+test_kernel = "Path"  # Define the kernel to use for the test image
+test_cmdline = "str"  # Define the kernel command line for the test image
+test_image_size = "int"  # Define the size of the test image in MiB
+test_memory = "str"  # Define the amount of memory to use for the test image (passed to qemu)
+test_cpu = "str"  # Define the CPU to use for the test image (passed to qemu)
+qemu_arch = "str"  # Define the qemu arch (added to the qemu-system- command)
+mounts = "dict"  # We only need the mounts dict, not the whole module
+test_flag = "str"  # Define the success flag used to determine if the test was successful
+test_timeout = "int"  # Define the timeout for the test
+

src/ugrd/generator_helpers.py

@@ -1,10 +1,10 @@
 from typing import Union
 from pathlib import Path
-from subprocess import run, CompletedProcess
+from subprocess import run, CompletedProcess, TimeoutExpired
 
 from zenlib.util import pretty_print
 
-__version__ = "1.0.0"
+__version__ = "1.1.0"
 __author__ = "desultory"
 
 
@@ -142,10 +142,15 @@ def _symlink(self, source: Union[Path, str], target: Union[Path, str]) -> None:
         self.logger.debug("Creating symlink: %s -> %s" % (target, source))
         symlink(source, target)
 
-    def _run(self, args: list[str]) -> CompletedProcess:
+    def _run(self, args: list[str], timeout=15) -> CompletedProcess:
         """ Runs a command, returns the CompletedProcess object """
-        self.logger.debug("Running command: %s" % ' '.join(args))
-        cmd = run(args, capture_output=True)
+        cmd_args = [str(arg) for arg in args]
+        self.logger.debug("Running command: %s" % ' '.join(cmd_args))
+        try:
+            cmd = run(cmd_args, capture_output=True, timeout=timeout)
+        except TimeoutExpired as e:
+            raise RuntimeError("[%ds] Command timed out: %s" % (timeout, [str(arg) for arg in cmd_args])) from e
+
         if cmd.returncode != 0:
             self.logger.error("Failed to run command: %s" % cmd.args)
             self.logger.error("Command output: %s" % cmd.stdout.decode())

src/ugrd/initramfs_dict.py

@@ -1,6 +1,6 @@
 
 __author__ = "desultory"
-__version__ = "1.5.2"
+__version__ = "1.7.0"
 
 from tomllib import load, TOMLDecodeError
 from pathlib import Path
@@ -39,11 +39,16 @@ def __init__(self, NO_BASE=False, *args, **kwargs):
                 super().__setitem__(parameter, default_type())
         if not NO_BASE:
             self['modules'] = 'ugrd.base.base'
+        else:
+            self['modules'] = 'ugrd.base.core'
 
     def import_args(self, args: dict) -> None:
         """ Imports data from an argument dict. """
         for arg, value in args.items():
             self.logger.info("Importing argument '%s' with value: %s" % (arg, value))
+            if arg == 'modules':
+                for module in value.split(','):
+                    self[arg] = module
             self[arg] = value
 
     def __setitem__(self, key: str, value) -> None:
@@ -119,6 +124,7 @@ def _process_custom_parameters(self, parameter_name: str, parameter_type: type)
 
         If the parameter is in the processing queue, process the queued values.
         """
+        from pycpio import PyCPIO
         self['custom_parameters'][parameter_name] = eval(parameter_type)
         self.logger.debug("Registered custom parameter '%s' with type: %s" % (parameter_name, parameter_type))
 
@@ -133,6 +139,8 @@ def _process_custom_parameters(self, parameter_name: str, parameter_type: type)
                 super().__setitem__(parameter_name, 0)
             case "float":
                 super().__setitem__(parameter_name, 0.0)
+            case "PyCPIO":
+                super().__setitem__(parameter_name, PyCPIO(logger=self.logger, _log_init=False, _log_bump=10))
             case _:  # For strings and things, don't init them so they are None
                 self.logger.debug("Leaving '%s' as None" % parameter_name)
 

src/ugrd/initramfs_generator.py

@@ -5,19 +5,16 @@
 
 from ugrd.initramfs_dict import InitramfsConfigDict
 
-from importlib.metadata import version
-
 from .generator_helpers import GeneratorHelpers
 
-__version__ = version(__package__)
 __author__ = "desultory"
 
 
 @loggify
 class InitramfsGenerator(GeneratorHelpers):
     def __init__(self, config='/etc/ugrd/config.toml', *args, **kwargs):
         self.config_filename = config
-        self.config_dict = InitramfsConfigDict(logger=self.logger)
+        self.config_dict = InitramfsConfigDict(NO_BASE=kwargs.pop('NO_BASE', False), logger=self.logger)
 
         # Used for functions that are added to the bash source file
         self.included_functions = {}
@@ -28,8 +25,11 @@ def __init__(self, config='/etc/ugrd/config.toml', *args, **kwargs):
         # init_pre and init_final are run as part of generate_initramfs_main
         self.init_types = ['init_debug', 'init_early', 'init_main', 'init_late', 'init_premount', 'init_mount', 'init_mount_late', 'init_cleanup']
 
-        self.load_config()
         self.config_dict.import_args(kwargs)
+        if config:
+            self.load_config()
+        else:
+            self.logger.warning("No config file specified, using the base config")
         self.config_dict.validate()
 
     def load_config(self) -> None:
@@ -127,11 +127,14 @@ def run_init_hook(self, level: str) -> list[str]:
             return out
         else:
             self.logger.debug("No output for init level: %s" % level)
+            return []
 
     def generate_profile(self) -> None:
         """ Generates the bash profile file based on self.included_functions. """
-        out = [self['shebang'],
-               f"#\n# Generated by UGRD v{__version__}"]
+        from importlib.metadata import version
+        ver = version(__package__) or 9999   # Version won't be found unless the package is installed
+        out = [self['shebang'].split(' ')[0],  # Don't add arguments to the shebang (for the profile)
+               f"#\n# Generated by UGRD v{ver}\n#"]
 
         # Add the library paths
         library_paths = ":".join(self['library_paths'])
@@ -172,7 +175,7 @@ def generate_init(self) -> None:
         self.run_hook('functions', force_include=True)
 
         init.extend(self.run_init_hook('init_pre'))
-        init += [f'einfo "Starting UGRD v{__version__}"']
+        init += [self.banner]
 
         if self['imports'].get('custom_init') and self.get('_custom_init_file'):
             init += ["\n# !!custom_init"]
@@ -191,7 +194,7 @@ def generate_init(self) -> None:
             self._write('/etc/profile', self.generate_profile(), 0o755)
             self.logger.info("Included functions: %s" % ', '.join(list(self.included_functions.keys())))
             if self['imports'].get('custom_init'):
-                custom_init.insert(2, f"einfo 'Starting custom init, UGRD v{__version__}'")
+                custom_init.insert(2, self.banner)
 
         if self.get('_custom_init_file'):
             self._write(self['_custom_init_file'], custom_init, 0o755)