tests/storage/linstor: Add test for forgetting and introducing Linstor SR

- Introduced `test_forget_and_introduce_sr` in `test_linstor_sr.py`:
  - Validates the ability to forget an SR and properly reintroduce it.
  - Ensures all PBD configurations are stored and restored correctly.
  - Checks that exceptions are raised when querying a forgotten SR.
  - Restores the SR and verifies successful reattachment of PBDs.

- Enhanced `SR` class in `sr.py` with:
  - `param_get()`: Fetch a specific SR parameter.
  - `type()`: Retrieve SR type.
  - `forget()`: Remove SR from the pool.
  - `introduce()`: Introduce SR.

- Improved `Host` command execution in `host.py`:
  - Added support for handling both dictionary and list-based arguments.
  - Fixed command formatting for `config:key=value` scenarios.

Signed-off-by: Rushikesh Jadhav <[email protected]>

Author: rushikeshjadhav

lib/host.py

@@ -86,8 +86,14 @@ def stringify(key, value):
                 return ret.rstrip()
             return "{}={}".format(key, shlex.quote(value))
 
-        command = ['xe', action] + maybe_param_minimal + maybe_param_force + \
-                  [stringify(key, value) for key, value in args.items()]
+        # Handle scenarios like config:key=value
+        # device-config:redundancy=2 device-config:provisioning=thin
+        command = ['xe', action] + maybe_param_minimal + maybe_param_force
+        if isinstance(args, dict):
+            command += [stringify(key, value) for key, value in args.items()]
+        elif isinstance(args, list):
+            command += [stringify(key, value) for key, value in args]
+
         result = self.ssh(
             command,
             check=check,

lib/sr.py

@@ -40,6 +40,7 @@ def unplug_pbds(self, force=False):
 
     def all_pbds_attached(self):
         all_attached = True
+        # TBD: handle sr that does not have PBD, following will return all_attached=True
         for pbd_uuid in self.pbd_uuids():
             all_attached = all_attached and self.pool.master.xe('pbd-param-get', {'uuid': pbd_uuid,
                                                                 'param-name': 'currently-attached'})
@@ -145,6 +146,20 @@ def main_host(self):
     def content_type(self):
         return self.pool.master.xe('sr-param-get', {'uuid': self.uuid, 'param-name': 'content-type'})
 
+    def param_get(self, param):
+        return self.pool.master.xe('sr-param-get', {'uuid': self.uuid, 'param-name': param})
+
+    def type(self):
+        return self.pool.master.xe('sr-param-get', {'uuid': self.uuid, 'param-name': 'type'})
+
+    def forget(self):
+        return self.pool.master.xe('sr-forget', {'uuid': self.uuid})
+
+    def introduce(self, type, shared, name_label, uuid):
+        return self.pool.master.xe('sr-introduce', {'uuid': self.uuid, 'type': type,
+                                                    'shared': shared, 'content-type': 'user',
+                                                    'name-label': name_label, 'uuid': uuid})
+
     def is_shared(self):
         if self._is_shared is None:
             self._is_shared = self.pool.master.xe('sr-param-get', {'uuid': self.uuid, 'param-name': 'shared'})

tests/storage/linstor/conftest.py

@@ -93,7 +93,7 @@ def linstor_sr(pool_with_linstor, provisioning_type, storage_pool_name):
         'provisioning': provisioning_type
     }, shared=True)
     yield sr
-    sr.destroy()
+    sr.destroy(verify=True)
 
 @pytest.fixture(scope='module')
 def vdi_on_linstor_sr(linstor_sr):

tests/storage/linstor/test_linstor_sr.py

@@ -4,7 +4,7 @@
 
 from .conftest import LINSTOR_PACKAGE
 from lib.commands import SSHCommandFailed
-from lib.common import wait_for, vm_image
+from lib.common import wait_for, vm_image, safe_split
 from tests.storage import vdi_is_open
 
 # Requirements:
@@ -78,6 +78,57 @@ def test_snapshot(self, vm_on_linstor_sr):
         finally:
             vm.shutdown(verify=True)
 
+    def test_forget_and_introduce_sr(self, linstor_sr):
+        from lib.sr import SR
+
+        sr = linstor_sr
+        sr_name = sr.param_get('name-label')
+        all_pbds = sr.pbd_uuids()
+        pbd_config_hosts = []
+        pbd_config_devices = []
+        # TBD: Move the pbd-param-get to either sr.py or introduce pbd.py
+        for pbd in all_pbds:
+            pbd_config_hosts.append(
+                safe_split(sr.pool.master.xe('pbd-param-get', {'uuid': pbd, 'param-name': 'host-uuid'})))
+            pbd_config_devices.append(
+                safe_split(sr.pool.master.xe('pbd-param-get', {'uuid': pbd, 'param-name': 'device-config'})))
+
+        if sr.all_pbds_attached():
+            sr.unplug_pbds()
+
+        sr.forget()
+        logging.info(f"Forgot SR {sr.uuid} successfully.")
+
+        with pytest.raises(Exception):
+            sr_type = sr.param_get('type') # Expecting exception as sr should not exist
+            sr.plug_pbds() # Plug back pbds and let teardown handle SR destroy
+            pytest.fail(f"SR still exists; returned type: {sr_type}")
+
+        logging.info(f"Introducing SR {sr.uuid} back.")
+        new_sr = sr.introduce(type='linstor', shared='true', name_label=sr_name, uuid=sr.uuid)
+
+        # Example pbd_config_device
+        # {provisioning: thin; redundancy: 3; group-name: linstor_group/thin_device}
+        for pbd_config_host, pbd_config_device in zip(pbd_config_hosts, pbd_config_devices):
+            pbd_config_dict = dict(
+                (kv.split(": ")[0].strip(), kv.split(": ")[1].strip())
+                for kv in pbd_config_device[0].split(";") if ": " in kv # Ensure key-value pair
+            )
+            device_config_entries = [('device-config:' + k, v) for k, v in pbd_config_dict.items()]
+
+            sr.pool.master.xe(
+                'pbd-create',
+                [
+                    ('sr-uuid', new_sr),
+                    ('host-uuid', pbd_config_host[0]),
+                    ('content-type', 'user'),
+                ] + device_config_entries
+            )
+
+        restored_sr = SR(new_sr, sr.pool)
+        restored_sr.plug_pbds(verify=True)
+        logging.info(f"Introduced SR {sr.uuid} successfully.")
+
     # *** tests with reboots (longer tests).
 
     @pytest.mark.reboot