Hibernation/Resume testing

src/ugrd/base/test.py

@@ -55,7 +55,7 @@ def _get_qemu_cmd_args(self, test_image):
         "-cpu": self["test_cpu"],
         "-kernel": self["test_kernel"],
         "-initrd": test_initrd,
-        "-append": self["test_cmdline"],
+        "-append": " ".join(self["test_cmdline"]),
         "-drive": "file=%s,format=raw" % test_rootfs,
     }
 
@@ -86,7 +86,7 @@ def make_test_image(self):
         "validate": False,
         "NO_BASE": True,
         "config": None,
-        "modules": "ugrd.fs.test_image",
+        "modules": ",".join(self["test_modules"]),  # By default is only "ugrd.fs.test_image"
         "out_file": self["test_rootfs_name"],
         "build_dir": self["test_rootfs_build_dir"],
         "custom_parameters": get_copy_config_types(self),
@@ -163,7 +163,10 @@ def test_image(self):
             if self["test_flag"] in line:
                 self.logger.info("Test flag found in output: %s", c_(line, "green"))
                 break
-            elif line.endswith("exitcode=0x00000000"):
+            elif (
+                line.endswith("exitcode=0x00000000")
+                or "---[ end Kernel panic - not syncing: sysrq triggered crash ]---" in line
+            ):
                 failed = True
                 break
             elif "press space" in line.lower():

src/ugrd/base/test.toml

@@ -1,16 +1,16 @@
 test_rootfs_name = 'ugrd-test-rootfs'
 test_rootfs_build_dir = 'initramfs_test_rootfs'
-test_image_size = 16
+test_image_size = 256
 
 test_copy_config = ["_mounts", "mounts", "out_dir", "tmpdir", "clean", "test_image_size", "test_no_rootfs", "test_flag", "cryptsetup"]
 kmod_init = ["ata_piix", "sd_mod"]
 
 test_memory = '256M'
 test_cpu = 'host'
 test_arch = 'x86_64'
-test_timeout = 15
-test_cmdline = 'console=ttyS0,115200 panic=1'
-#qemu_bool_args = ['nographic', 'no-reboot', 'enable-kvm']
+test_timeout = 30
+test_cmdline = ['console=ttyS0,115200', 'panic=0']
+test_modules = [ "ugrd.fs.test_image" ]
 qemu_bool_args = ['nographic', 'enable-kvm' ]
 
 [imports.build_pre]
@@ -24,11 +24,12 @@ test_kernel = "Path"  # Define the kernel to use for the test
 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)
 test_arch = "str"  # Define the qemu arch (added to the qemu-system- command)
-test_cmdline = "str"  # Define the kernel command line for the test image
+test_cmdline = "NoDupFlatList"  # Define the kernel command line for the test image
 test_timeout = "int"  # Define the timeout for the test
 test_image_size = "int"  # Define the size of the test image, in MB
 test_flag = "str"  # Define the success flag for the test
 test_no_rootfs = "bool"  # Toggle to run tests without mounting the root image
 test_rootfs_name = "str"  # Define the name of the rootfs image
 test_rootfs_build_dir = "Path"  # Define the build directory for the rootfs image
+test_modules = "NoDupFlatList" # List of modules to include in test image
 qemu_bool_args = "NoDupFlatList"  # Define the qemu boolean arguments

src/ugrd/fs/resume.py

@@ -3,6 +3,22 @@
 from zenlib.util import contains
 
 
+@contains("test_resume")
+def setup_resume_tests(self) -> None:
+    if "ugrd.base.test" in self["modules"]:
+        from uuid import uuid4
+
+        # Create a uuid for the swap partition in the test image
+        if not self["test_swap_uuid"]:
+            self["test_swap_uuid"] = uuid4()
+
+        # pull in the hibernation/resume testing module
+        self["test_modules"] = "ugrd.fs.test_resume"
+
+        # append resume partition to QEMU kernel cmdline
+        self["test_cmdline"] = f"resume=UUID={self['test_swap_uuid']}"
+
+
 def resume(self) -> str:
     """Returns a shell script handling resume from hibernation.
     Checks that /sys/power/resume is writable, resume= is set, and noresume is not set, if so,
@@ -27,10 +43,10 @@ def resume(self) -> str:
             eerror "Cannot safely resume with mounted block devices:\n$(lsblk -Q MOUNTPOINT -no PATH)"
             return 1
         fi
+
         [ -b "$1" ] || (ewarn "\'$1\' is not a valid block device!" ; return 1)
         einfo "Attempting resume from: $1"
-        # TODO: This could maybe be printf?
-        echo -n "$1" > /sys/power/resume
+        printf "%s" "$1" > /sys/power/resume
         einfo "No image on: $resume"
         return 0
     """
@@ -66,18 +82,6 @@ def handle_late_resume(self) -> None:
     self.logger.warning(
         "[late_resume] enabled, this can result in data loss if filesystems are modified before resuming. Read the docs for more info."
     )
-    return handle_early_resume(
-        self
-    )  # At the moment it's the same code but delayed, will change when more features are added
-
-
-@contains("test_resume")
-def test_init_swap_uuid(self):
-    if "test_cpu" in self:
-        from uuid import uuid4
-
-        self["test_swap_uuid"] = swap_uuid = uuid4()
 
-        # append to test kernel cmdline and adjust planned image size to allow enough space
-        self["test_cmdline"] = f"{self.get('test_cmdline')} resume=UUID={swap_uuid}"
-        self["test_image_size"] = 256 + self.get("test_image_size")
+    # At the moment it's the same code but delayed, will change when more features are added
+    return handle_early_resume(self)

src/ugrd/fs/resume.toml

@@ -3,7 +3,7 @@ binaries = [ 'lsblk' ]
 test_copy_config = [ "test_resume", "test_swap_uuid" ]
 
 [imports.build_pre]
-"ugrd.fs.resume" = [ "test_init_swap_uuid" ]
+"ugrd.fs.resume" = [ "setup_resume_tests" ]
 
 [imports.init_main]
 "ugrd.fs.resume" = [ "handle_early_resume", "handle_late_resume"]
@@ -15,10 +15,9 @@ handle_early_resume = "mount_fstab"
 handle_late_resume = ["crypt_init", "init_lvm"]
 
 [imports.functions]
-"ugrd.fs.resume" = ["resume"]
+"ugrd.fs.resume" = [ "resume" ]
 
 [custom_parameters]
-late_resume = "bool"
-test_resume = "bool"
-test_swap_uuid = "str"
-
+late_resume = "bool"  # Include code for late resume
+test_resume = "bool" # Test hibernation/resume pathways when test is enabled
+test_swap_uuid = "str" # Define the uuid of the swap partition on the test image

src/ugrd/fs/test_image.py

@@ -9,37 +9,19 @@
 
 MIN_FS_SIZES = {"btrfs": 110, "f2fs": 50}
 
-@contains("test_flag", "A test flag must be set to create a test image", raise_exception=True)
+
 def init_banner(self):
     """Initialize the test image banner, set a random flag if not set."""
-    self["banner"] = f"echo {self['test_flag']}"
-
-
-@contains("test_resume")
-def resume_tests(self):
-    return [
-        'if [ "$(</sys/power/resume)" != "0:0" ] ; then',
-        '   [ -e "/resumed" ] && (rm /resumed ; echo c > /proc/sysrq-trigger)',
-        # Set correct resume parameters
-        "   echo reboot > /sys/power/disk",
-        # trigger resume
-        "   echo disk > /sys/power/state",
-        '   [ -e "/resume" ] || echo c > /proc/sysrq-trigger',
-        # if we reach this point, resume was successful
-        # reset environment in case resume needs to be rerun
-        "   rm /resumed",
-        '   echo "Resume completed without error.',
-        "else",
-        '   echo "No resume device found! Resume test not possible!',
-        "fi",
-    ]
+    self["banner"] = "echo ugRD Test Image"
 
 
+@contains("test_flag", "A test flag must be set to create a test image", raise_exception=True)
 def complete_tests(self):
-    return [
-        "echo s > /proc/sysrq-trigger",
-        "echo o > /proc/sysrq-trigger",
-    ]
+    return f"""
+        echo {self["test_flag"]}
+        echo s > /proc/sysrq-trigger
+        echo o > /proc/sysrq-trigger
+    """
 
 
 def _allocate_image(self, image_path, padding=0):
@@ -61,12 +43,13 @@ def _allocate_image(self, image_path, padding=0):
 
     with open(image_path, "wb") as f:
         total_size = (self.test_image_size + padding) * (2**20)  # Convert MB to bytes
-        self.logger.info(f"Allocating {self.test_image_size + padding}MB test image file: { c_(f.name, 'green')}")
-        self.logger.debug(f"[{f.name}] Total bytes: { c_(total_size, 'green')}")
+        self.logger.info(f"Allocating {self.test_image_size + padding}MB test image file: {c_(f.name, 'green')}")
+        self.logger.debug(f"[{f.name}] Total bytes: {c_(total_size, 'green')}")
         f.write(b"\0" * total_size)
 
+
 def _copy_fs_contents(self, image_path, build_dir):
-    """ Mount and copy the filesystem contents into the image,
+    """Mount and copy the filesystem contents into the image,
     for filesystems which cannot be created directly from a directory"""
     try:
         with TemporaryDirectory() as tmp_dir:
@@ -149,11 +132,9 @@ def make_test_image(self):
         _allocate_image(self, image_path)
 
     loopback = None
-    if self.get("test_resume"):
+    if self.get("test_swap_uuid"):
         try:
-            self._run(["sgdisk", "-og", image_path])
-            self._run(["sgdisk", "-n", "1:0:+256", image_path])
-            self._run(["sgdisk", "-n", "2:0", image_path])
+            self._run(["sgdisk", "-og", "-n", "1:0:+128M", "-n", "2:0:0", image_path])
         except RuntimeError as e:
             raise RuntimeError("Failed to partition test disk: %s", e)
 
@@ -163,6 +144,7 @@ def make_test_image(self):
 
             image_path = f"{loopback}p2"
         except RuntimeError as e:
+            self._run(["losetup", "-d", loopback])  # Free loopback device on fail
             raise RuntimeError("Failed to allocate loopback device for disk creation: %s", e)
 
         # sleep for 100ms, to give the loopback device time to scan for partitions
@@ -173,6 +155,7 @@ def make_test_image(self):
         try:
             self._run(["mkswap", "-U", self["test_swap_uuid"], f"{loopback}p1"])
         except RuntimeError as e:
+            self._run(["losetup", "-d", loopback])
             raise RuntimeError("Failed to create swap partition on test disk: %s", e)
 
     if rootfs_type == "ext4":

src/ugrd/fs/test_image.toml

@@ -1,17 +1,15 @@
 modules = [ 'ugrd.base.banner' ]  # get_archive_path
 paths = ['dev', 'sys', 'proc']
+binaries = ['swapon']
 
-test_image_size = 16
-shebang = "#!/bin/sh"  # Don't use -l, no profile is generated
+test_image_size = 256
+shebang = "#!/bin/sh -l"
 _cryptsetup_root = "root"
 
 
 [imports.build_pre]
 "ugrd.fs.test_image" = ["init_banner"]
 
-[imports.init_main]
-"ugrd.fs.test_image" = ["resume_tests"]
-
 [imports.init_final]
 "ugrd.fs.test_image" = ["complete_tests"]
 
@@ -28,3 +26,4 @@ test_image_size = "int"  # Define the size of the test image in MiB
 test_flag = "str"  # Define the success flag used to determine if the test was successful
 check_included_or_mounted = "NoDupFlatList"  # Only defined to suppress errors about unknown config
 test_resume = "bool" # Enable code to test the suspend/resume pathways
+test_swap_uuid = "str" # Define the UUID of the swap partition

src/ugrd/fs/test_resume.py

@@ -0,0 +1,26 @@
+__version__ = "2.0.0"
+
+from zenlib.util import contains
+
+
+@contains("test_resume")
+def resume_tests(self):
+    return """
+        echo "Begin resume testing."
+        if [ "$(</sys/power/resume)" != "0:0" ] ; then
+            echo reboot > /sys/power/disk
+            echo 1 > /sys/power/pm_debug_messages
+
+            echo "Activating hibernation swap device..."
+            swapon /dev/$(. "/sys/dev/block/$(read blk </sys/power/resume && echo $blk)/uevent" && echo $DEVNAME)
+
+            echo "Triggering test hibernation..."
+            echo disk > /sys/power/state || (echo "Suspend to disk failed!" ; echo c > /proc/sysrq-trigger)
+
+            # Assume at this point system has hibernated then resumed again
+            echo "Resume test completed without error."
+        else
+            echo "No resume device found! Resume test not possible!"
+            echo c > /proc/sysrq-trigger
+        fi
+    """

src/ugrd/fs/test_resume.toml

@@ -0,0 +1,5 @@
+[imports.init_main]
+"ugrd.fs.test_resume" = ["resume_tests"]
+
+[custom_parameters]
+test_resume = "bool"

tests/resume.toml

@@ -0,0 +1,18 @@
+# Hibernation/Resume tests
+modules = [
+	"ugrd.fs.resume",
+	"ugrd.base.test",
+]
+
+late_resume = true
+test_resume = true
+
+out_dir = "initramfs_test"
+test_image_size = 256
+
+cpio_compression = false
+hostonly = false
+
+[mounts.root]
+uuid = "aaaabbbb-cccc-dddd-eeee-ffff00000000"
+type = "ext4"

tests/test_resume.py

@@ -0,0 +1,13 @@
+from unittest import TestCase, main
+
+from ugrd.initramfs_generator import InitramfsGenerator
+from zenlib.logging import loggify
+
+@loggify
+class TestResume(TestCase):
+    def test_resume(self):
+        generator = InitramfsGenerator(logger=self.logger, config="tests/resume.toml")
+        generator.build()
+
+if __name__ == "__main__":
+    main()