Optest sysinfo enhance and Dlpar mem_hotplug testcase

CONFIG.YAML

@@ -15,3 +15,17 @@ HMC:
        - "lshwres -r virtualio --rsubtype eth -m SYS --level lpar" #Virtual Ethernet Adapter (VEA) Info
        - "lshwres -r virtualio --rsubtype scsi -m SYS --level lpar" #scsi Adapter Info
        - "lshwres -r virtualio --rsubtype fc -m SYS --level lpar"  #fc Adapter Info
+
+
+IO:
+  COMMANDS:
+    - "lshwres -r io -m SYS --rsubtype slot --filter 'lpar_names=LPAR_NAME'"  # List physical IO
+    - "lshwres -r virtualio --rsubtype fc --level lpar -m SYS -F adapter_type,lpar_name,slot_num,remote_lpar_name,remote_slot_num,wwpns"  # List virtual IOs
+
+KVM:
+  COMMANDS:
+    - "arch"  # Architecture
+    - "lscpu | grep 'Model name' | cut -f 2 -d ':' | awk '{$1=$1}1'"  # Processor
+    - "lsmcode"  # Firmware
+    - "nproc"  # No-of-CPUs
+    - "free -mh"  # RAM

common/OpTestHMC.py

@@ -1141,6 +1141,10 @@ def set_system(self, system):
         self.sysinfo.get_OSconfig(self.pty, self.expect_prompt)
         log.info("Collecting HMC details")
         self.sysinfo.get_HMCconfig(self.ssh, self.expect_prompt,self.mg_system)
+        log.info("Collecting IO details")
+        self.sysinfo.get_IOconfig(self.ssh, self.expect_prompt,self.mg_system,self.lpar_name)
+        log.info("Collecting KVM info")
+        self.sysinfo.get_KVMconfig(self.pty, self.expect_prompt)
 
     def get_host_console(self):
         '''

common/OpTestInstallUtil.py

@@ -466,9 +466,14 @@ def update_kernel_cmdline(self, distro, args="", remove_args="", reboot=True,
         if reboot and (req_args or req_remove_args):
             # Reboot the host for the kernel command to reflect
             if reboot_cmd:
+                # Always reopen console fresh to avoid stale session after first reboot
+                self.cv_SYSTEM.console.close()
                 raw_pty = self.cv_SYSTEM.console.get_console()
                 raw_pty.sendline("reboot")
-                raw_pty.expect("login:", timeout=900)
+                login_patterns = [
+                  "login:", "root login:", "Ubuntu login:", "Password:",
+                ]
+                raw_pty.expect(login_patterns, timeout=900)
             else:
                 self.cv_SYSTEM.goto_state(OpSystemState.OFF)
                 self.cv_SYSTEM.goto_state(OpSystemState.OS)

common/OpTestSOL.py

@@ -93,6 +93,14 @@ def nontimeout_run(self):
                 self.c.expect("\n", timeout=60)
             except pexpect.TIMEOUT:
                 pass
+            except OSError as e:
+                # File descriptor closed (reboot / rmvterm)
+                print(f"[OpTestSOL] Console closed: {e}, stopping SOL thread.")
+                break
+            except Exception as e:
+                # Catch-all for any other console errors
+                print(f"[OpTestSOL] Unexpected error in SOL thread: {e}, stopping.")
+                break
             except pexpect.EOF:
                 self.c.close()
                 self.c = self.system.console.get_console()

common/OpTestSysinfo.py

@@ -42,6 +42,8 @@ def __init__(self):
             self.config_actions = yaml.safe_load(file)
             list_of_commands = self.config_actions["LINUX"]["COMMANDS"]
             get_HMCconfig_cmds  = self.config_actions["HMC"]["COMMANDS"]
+            get_IOconfig_cmds  = self.config_actions["IO"]["COMMANDS"]
+            get_KVMconfig_cmds  = self.config_actions["KVM"]["COMMANDS"]
 
     def get_OSconfig(self, pty, prompt):
         # Collect config related data from the OS
@@ -70,4 +72,33 @@ def get_HMCconfig(self, pty, prompt,CEC_name):
                     output = pty.run_command(each_cmd)
                 except Exception as e:
                           print('command failed due to system error')
-
+    def get_IOconfig(self, pty, prompt,CEC_name,LPAR):
+        # Collect config data from HMC
+        ################ IO INFO ####################
+        #lpar_name = pty.run_command("lparstat -i | awk -F: '/Partition Name/ {print $2}' | xargs").strip()
+        get_HMCconfig_cmds  = self.config_actions["IO"]["COMMANDS"]
+        for index, each_cmd in enumerate(get_HMCconfig_cmds, start=0):
+             if re.search(r'SYS|LPAR_NAME', each_cmd):
+                new_cmd = each_cmd
+                # Replace placeholders
+                new_cmd = re.sub(r'SYS', CEC_name, new_cmd)
+                new_cmd = re.sub(r'LPAR_NAME', LPAR, new_cmd)
+                try:
+                   output = pty.run_command(new_cmd)
+                except Exception as e:
+                   print('command failed due to system error')
+             else:
+                try:
+                    output = pty.run_command(each_cmd)
+                except Exception as e:
+                          print('command failed due to system error') 
+    def get_KVMconfig(self, pty, prompt):
+        # Collect config related data from the OS
+        try:
+            list_of_commands = self.config_actions["KVM"]["COMMANDS"]
+            print("########### KVM info ########")
+            for index, each_cmd in enumerate(list_of_commands, start=0):
+                pty.sendline(each_cmd)
+                rc = pty.expect([prompt, pexpect.TIMEOUT, pexpect.EOF], timeout=10)
+        except CommandFailed as cf:
+            raise cf 

testcases/OpTestDlpar.py

@@ -38,23 +38,30 @@
 lpar2_name - name of destination lpar for move operation
 loop_num - number of times to run loop
 '''
+import time
 import unittest
 import logging
 import os.path
 from os import path 
 import OpTestConfiguration
 import OpTestLogger
+from testcases.grub import Grub
 from common import OpTestHMC, OpTestFSP
 from common import OpTestHMC
 from common.OpTestSystem import OpSystemState
 from common.OpTestConstants import OpTestConstants as BMC_CONST
 from random import randint
 from common.OpTestSystem import OpSystemState
 from common.OpTestSOL import OpSOLMonitorThread
+from common import OpTestInstallUtil
+from common.OpTestUtil import OpTestUtil
+
 log = OpTestLogger.optest_logger_glob.get_logger(__name__)
 class OpTestDlpar(unittest.TestCase):
    def setUp(self):
       conf = OpTestConfiguration.conf
+      self.op_test_util = OpTestUtil(conf)
+      self.distro = self.op_test_util.distro_name()
       self.cv_SYSTEM = conf.system()
       self.console = self.cv_SYSTEM.console
       conf = OpTestConfiguration.conf
@@ -315,6 +322,43 @@ def runTest(self):
             log.debug("Deleting smt script")
             self.console.run_command("rm ./smt_script")
 
+
+class Dlpar_mem_hotplug(OpTestDlpar, unittest.TestCase):
+    """Class for DLPAR Memory hotplug Tests
+       This class executes test cases from OpTestDlpar.DlparMemBasic.
+
+       Step 1 : memory_hotplug.memmap_on_memory=0 * test dlpar memory add, memory remove
+       Step 2 : memory_hotplug.memmap_on_memory=1 * test dlpar memory add, memory remove
+       Step 3 : memory_hotplug.memmap_on_memory=force * test dlpar memory add, memory remove
+    """
+
+    def setUp(self):
+        super(Dlpar_mem_hotplug, self).setUp()
+
+    def runTest(self):
+        obj = OpTestInstallUtil.InstallUtil()
+
+        test_settings = [0, 1, 'force']
+        for setting in test_settings:
+            obj.update_kernel_cmdline(
+                self.distro,
+                args=f"memory_hotplug.memmap_on_memory={setting}",
+                reboot=True,
+                reboot_cmd=True
+            )
+            # Establish SSH connection
+            con = self.cv_SYSTEM.cv_HOST.get_ssh_connection()
+
+            log.debug("Memory hotplug with setting: %s", setting)
+            log.debug("=================")
+            # Add memory resource
+            self.AddRemove("mem", "-q", "a", self.mem_resource)
+
+            #log.debug("Memory hotplug removal with setting: %s", setting)
+            log.debug("=================")
+            # Remove memory resource
+            self.AddRemove("mem", "-q", "r", self.mem_resource)
+
 def tearDown(self):
     self.console_thread.console_terminate()
 #reboot machine & delete script smt_script