fix(coalesce): Refactor function to deduplicate

Signed-off-by: Damien Thenot <[email protected]>

Author: Nambrok

tests/storage/coalesce/conftest.py

@@ -24,7 +24,7 @@ def vdi_with_vbd_on_dom0(host, vdi_on_local_sr):
 
     dom0.disconnect_vdi(vdi_on_local_sr)
 
-@pytest.fixture(scope="class")
+@pytest.fixture(scope="function")
 def data_file_on_host(host):
     filename = "/root/data.bin"
     logging.info(f">> Creating data file {filename} on host")

tests/storage/coalesce/test_coalesce.py

@@ -1,51 +1,40 @@
-import logging
+import pytest
 
-from .utils import compare_data, copy_data_to_tapdev, snapshot_vdi, wait_for_vdi_coalesce
+import logging
 
-class Test:
-    def test_write_data(self, host, tapdev, data_file_on_host):
-        length = 1 * 1024 * 1024
-        offset = 0
+from .utils import compare_data, copy_data_to_tapdev, operation_on_vdi, wait_for_vdi_coalesce
 
-        logging.info("Copying data to tapdev")
-        copy_data_to_tapdev(host, data_file_on_host, tapdev, offset, length)
+from typing import TYPE_CHECKING
 
-        logging.info("Comparing data to tapdev")
-        assert compare_data(host, tapdev, data_file_on_host, offset, length)
+if TYPE_CHECKING:
+    from lib.host import Host
+    from lib.vdi import VDI
 
-    def test_coalesce(self, host, tapdev, vdi_with_vbd_on_dom0, data_file_on_host):
-        vdi = vdi_with_vbd_on_dom0
-        vdi_uuid = vdi.uuid
+class Test:
+    def test_write_data(self, host: "Host", tapdev: str, data_file_on_host: str):
         length = 1 * 1024 * 1024
         offset = 0
 
-        vdi_snap = snapshot_vdi(host, vdi_uuid)
-
         logging.info("Copying data to tapdev")
         copy_data_to_tapdev(host, data_file_on_host, tapdev, offset, length)
 
-        logging.info("Removing VDI snapshot")
-        host.xe("vdi-destroy", {"uuid": vdi_snap})
-
-        wait_for_vdi_coalesce(vdi)
-
         logging.info("Comparing data to tapdev")
         assert compare_data(host, tapdev, data_file_on_host, offset, length)
 
-    def test_clone_coalesce(self, host, tapdev, vdi_with_vbd_on_dom0, data_file_on_host):
+    @pytest.mark.parametrize("vdi_op", ["snapshot", "clone"])
+    def test_coalesce(self, host: "Host", tapdev: str, vdi_with_vbd_on_dom0: "VDI", data_file_on_host: str, vdi_op):
         vdi = vdi_with_vbd_on_dom0
         vdi_uuid = vdi.uuid
         length = 1 * 1024 * 1024
         offset = 0
 
-        clone_uuid = host.xe("vdi-clone", {"uuid": vdi_uuid})
-        logging.info(f"Clone VDI {vdi_uuid}: {clone_uuid}")
+        new_vdi = operation_on_vdi(host, vdi_uuid, vdi_op)
 
         logging.info("Copying data to tapdev")
         copy_data_to_tapdev(host, data_file_on_host, tapdev, offset, length)
 
-        logging.info("Removing VDI clone")
-        host.xe("vdi-destroy", {"uuid": clone_uuid})
+        logging.info(f"Removing VDI {vdi_op}")
+        host.xe("vdi-destroy", {"uuid": new_vdi})
 
         wait_for_vdi_coalesce(vdi)
 

tests/storage/coalesce/utils.py

@@ -4,6 +4,8 @@
 from lib.host import Host
 from lib.vdi import VDI
 
+from typing import Literal
+
 def wait_for_vdi_coalesce(vdi: VDI):
     wait_for_not(lambda: vdi.get_parent(), msg="Waiting for coalesce")
     logging.info("Coalesce done")
@@ -31,10 +33,10 @@ def get_data(host: Host, file: str, offset: int, length: int, checksum: bool = F
 def get_hashed_data(host: Host, file: str, offset: int, length: int):
     return get_data(host, file, offset, length, True).split()[0]
 
-def snapshot_vdi(host: Host, vdi_uuid: str):
-    vdi_snap = host.xe("vdi-snapshot", {"uuid": vdi_uuid})
-    logging.info(f"Snapshot VDI {vdi_uuid}: {vdi_snap}")
-    return vdi_snap
+def operation_on_vdi(host: Host, vdi_uuid: str, vdi_op: Literal["snapshot", "clone"]) -> str:
+    new_vdi = host.xe(f"vdi-{vdi_op}", {"uuid": vdi_uuid})
+    logging.info(f"{vdi_op.capitalize()} VDI {vdi_uuid}: {new_vdi}")
+    return new_vdi
 
 def compare_data(host: Host, tapdev: str, data_file: str, offset: int, length: int) -> bool:
     logging.info("Getting data from VDI and file")