OpTestKernelDump.py: Test crashkernel with memory offset parameter.

common/OpTestInstallUtil.py

@@ -405,6 +405,28 @@ def check_kernel_cmdline(self, args="", remove_args=""):
                    (Err.command, Err.output)))
         return req_args.strip(), req_remove_args.strip()
 
+    def update_or_replace_arg(self, cmdline, new_arg):
+
+        key = new_arg.split("=")[0]
+        parts = cmdline.split()
+
+        updated = []
+        replaced = False
+
+        for part in parts:
+            if part.startswith(key + "="):
+                if not replaced:
+                    updated.append(new_arg)
+                    replaced = True
+            # skip old duplicate
+            else:
+                updated.append(part)
+
+        if not replaced:
+            updated.append(new_arg)
+
+        return " ".join(updated)
+
     def update_kernel_cmdline(self, distro, args="", remove_args="", reboot=True,
                               reboot_cmd=False, timeout=0):
         """
@@ -442,7 +464,8 @@ def update_kernel_cmdline(self, distro, args="", remove_args="", reboot=True,
                 output = con.run_command(cmd, timeout=60)[0].replace("\"", "")
                 output = output.split("GRUB_CMDLINE_LINUX_DEFAULT=")[-1].strip()
                 if req_args:
-                    output += " %s" % req_args
+                    for arg in req_args.split():
+                        output = self.update_or_replace_arg(output, arg)
                 if req_remove_args:
                     for each_arg in req_remove_args.split():
                         output = output.replace(each_arg, "")

testcases/OpTestKernelDump.py

@@ -2176,6 +2176,78 @@ def runTest(self):
                 "%s: Dump Time = %.2f sec ( %.2f sec), VMCore Size = %.2f MB ( %.2f MB)" %
                 (name, data['time'], time_diff, data['vmcore_size_mb'], size_diff))
 
+class KernelCrash_CrashkernelOffset(OptestKernelDump):
+    '''
+    Test crashkernel with memory offset parameter.
+    Appends offset to existing crashkernel value and triggers crash.
+    Example: crashkernel=512M becomes crashkernel=512M@256M
+    '''
+
+    def runTest(self):
+        self.setup_test()
+        conf = OpTestConfiguration.conf
+
+        # Get test offset (default 256M)
+        try:
+            self.test_offset = conf.args.test_offset
+        except AttributeError:
+            self.test_offset = '256M'
+        log.info(f"========== Testing crashkernel with offset @{self.test_offset} ==========")
+
+        # Get current crashkernel value
+        cmdline = self.c.run_command("cat /proc/cmdline")
+        cmdline_str = ' '.join(cmdline)
+
+        match = re.search(r'crashkernel=([^\s@]+)', cmdline_str)
+        if not match:
+            self.skipTest("No crashkernel parameter found")
+
+        current_value = match.group(1)
+        log.info(f"Current crashkernel value: {current_value}")
+        old_param = f"crashkernel={current_value}"
+        # Create new parameter with offset
+        new_param = f"crashkernel={current_value}@{self.test_offset}"
+        log.info(f"New crashkernel parameter: {new_param}")
+        self.reset_kdump_bootloaded_if_needed() 
+        # Update using OpTestInstallUtil (will replace old crashkernel)
+        obj = OpTestInstallUtil.InstallUtil()
+        if not obj.update_kernel_cmdline(self.distro, 
+                                        args=new_param,
+                                        reboot=True, 
+                                        reboot_cmd=True):
+            self.fail("Failed to update crashkernel with offset")
+
+        # Verify crashkernel with offset
+        self.cv_SYSTEM.goto_state(OpSystemState.OFF)
+        self.cv_SYSTEM.goto_state(OpSystemState.OS)
+        cmdline = self.c.run_command("cat /proc/cmdline")
+        cmdline_str = ' '.join(cmdline)
+
+        if f"@{self.test_offset}" not in cmdline_str:
+            self.fail(f"Crashkernel offset @{self.test_offset} not found in kernel cmdline")
+
+        log.info(f"Verified: crashkernel with offset @{self.test_offset}")
+
+        # Trigger crash
+        log.info("========== Triggering crash ==========")
+        boot_type = self.kernel_crash()
+
+        # Verify dump file
+        self.verify_dump_file(boot_type)
+
+        log.info(f"========== SUCCESS: Crashkernel offset @{self.test_offset} test passed ==========")
+        old_param = f"crashkernel={current_value}"
+        log.info(f"old crashkernel parameter: {old_param}")
+        obj = OpTestInstallUtil.InstallUtil()
+        if not obj.update_kernel_cmdline(self.distro,
+                                        args=old_param,
+                                        remove_args=new_param,
+                                        reboot=True,
+                                        reboot_cmd=True):
+            self.fail("KernelArgTest failed to update kernel args")
+        self.cv_SYSTEM.goto_state(OpSystemState.OFF)
+        self.cv_SYSTEM.goto_state(OpSystemState.OS)
+
 
 def crash_suite():
     s = unittest.TestSuite()
@@ -2215,5 +2287,6 @@ def crash_suite():
     s.addTest(OPALCrash_MPIPL())
     s.addTest(PstoreCheck())
     s.addTest(MeasureMakedumpTime())
+    s.addTest(KernelCrash_CrashkernelOffset())
 
     return s