Merge pull request #279 from rushikeshjadhav/feat-storage-linstor-611

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

Author: klmp200

lib/host.py

@@ -43,6 +43,7 @@
 XAPI_CONF_FILE = '/etc/xapi.conf'
 XAPI_CONF_DIR = '/etc/xapi.conf.d'
 
+
 def host_data(hostname_or_ip: str) -> dict[str, str]:
     # read from data.py
     from data import HOST_DEFAULT_PASSWORD, HOST_DEFAULT_USER, HOSTS
@@ -171,7 +172,7 @@ def xe(self, action: str, args: dict[str, str | bool | dict[str, str]] = {}, *,
 
         def stringify(key: str, value: str | bool | dict[str, str]) -> str:
             if isinstance(value, bool):
-                return "{}={}".format(key, to_xapi_bool(value))
+                return f"{key}={to_xapi_bool(value)}"
             if isinstance(value, dict):
                 ret = ""
                 for key2, value2 in value.items():

lib/sr.py

@@ -20,7 +20,7 @@
 )
 from lib.vdi import VDI, ImageFormat
 
-from typing import TYPE_CHECKING, Literal, overload
+from typing import TYPE_CHECKING, Literal, Self, overload
 
 if TYPE_CHECKING:
     from lib.host import Host
@@ -202,6 +202,16 @@ def param_clear(self, param_name: str) -> None:
     def content_type(self) -> str:
         return self.param_get('content-type')
 
+    @classmethod
+    def introduce(cls, pool: Pool, type: str, shared: bool, name_label: str, sr_uuid: str) -> Self:
+        return cls(
+            pool.master.xe(
+                'sr-introduce',
+                {'uuid': sr_uuid, 'type': type, 'shared': shared, 'content-type': 'user', 'name-label': name_label},
+            ),
+            pool,
+        )
+
     def is_shared(self) -> bool:
         if self._is_shared is None:
             self._is_shared = strtobool(self.param_get('shared'))

tests/storage/linstor/test_linstor_sr.py

@@ -206,6 +206,53 @@ def test_resize_vdi(self, vm_on_linstor_sr: VM) -> None:
         vm.wait_for_os_booted()
         vm.shutdown(verify=True)
 
+    def test_forget_and_introduce_sr(self, linstor_sr: SR):
+        sr = linstor_sr
+        sr_name = sr.param_get('name-label')
+        all_pbds = sr.pbd_uuids()
+        pbd_config_hosts: list[list[str]] = []
+        pbd_config_devices: list[list[str]] = []
+        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(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(sr.pool, type='linstor', shared=True, name_label=sr_name, sr_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
+            )
+
+            sr.pool.master.xe(
+                "pbd-create",
+                {
+                    "sr-uuid": new_sr.uuid,
+                    "host-uuid": pbd_config_host[0],
+                    "content-type": "user",
+                    "device-config": pbd_config_dict,
+                },
+            )
+
+        new_sr.plug_pbds(verify=True)
+        logging.info(f"Introduced SR {new_sr.uuid} successfully")
+
     # *** tests with reboots (longer tests).
 
     @pytest.mark.reboot