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

lib/host.py

@@ -130,8 +130,16 @@ def stringify(key, value):
                 return ret.rstrip()
             return "{}={}".format(key, shlex.quote(value))
 
-        command: List[str] = ['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: List[str] = ['xe', action] + maybe_param_minimal + maybe_param_force
+        resolved_args = None
+        if isinstance(args, list):
+            resolved_args = args
+        else:
+            resolved_args = args.items()
+        command += [stringify(key, value) for key, value in resolved_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 strtobool(self.pool.master.xe('pbd-param-get',
                                                                           {'uuid': pbd_uuid,
@@ -145,7 +146,18 @@ def main_host(self):
         return self._main_host
 
     def content_type(self):
-        return self.pool.master.xe('sr-param-get', {'uuid': self.uuid, 'param-name': 'content-type'})
+        return self.param_get('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.param_get('type')
+
+    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:

tests/storage/linstor/conftest.py

@@ -94,7 +94,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:
@@ -86,6 +86,54 @@ 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'})))
+
+        sr.forget()
+        logging.info("Forgot SR %s successfully", sr.uuid)
+
+        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("Introducing SR %s back", sr.uuid)
+        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("Introduced SR %s successfully", sr.uuid)
+
     # *** tests with reboots (longer tests).
 
     @pytest.mark.reboot