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.

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

Author: rushikeshjadhav

Diff for: 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):

Diff for: 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