lib/basevm.py: add vdis related function

Add a list of vdis object to a VM
Add get_dom0_vm in host.py
Add function to connect/disconnect a VDI from a VM

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

Author: Nambrok

lib/basevm.py

@@ -1,12 +1,14 @@
 import logging
 
-from typing import TYPE_CHECKING, Any, Literal, Optional, overload
+from typing import TYPE_CHECKING, Any, List, Literal, Optional, overload
 
 if TYPE_CHECKING:
     import lib.host
 
+from lib.commands import SSHCommandFailed
 from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set
 from lib.sr import SR
+from lib.vdi import VDI
 
 class BaseVM:
     """ Base class for VM and Snapshot. """
@@ -18,6 +20,15 @@ def __init__(self, uuid: str, host: 'lib.host.Host'):
         logging.info("New %s: %s", type(self).__name__, uuid)
         self.uuid = uuid
         self.host = host
+        try:
+            self.vdis = [VDI(vdi_uuid, host=host) for vdi_uuid in self.vdi_uuids()]
+        except SSHCommandFailed as e:
+            # Doesn't work with Dom0 since `vm-disk-list` doesn't work on it so we create empty list
+            if e.stdout == "Error: No matching VMs found":
+                logging.info("Couldn't get disks list. We are Dom0. Continuing...")
+                self.vdis = []
+            else:
+                raise
 
     @overload
     def param_get(self, param_name: str, key: Optional[str] = ...,
@@ -59,7 +70,7 @@ def name(self) -> str:
     def _disk_list(self):
         raise NotImplementedError()
 
-    def vdi_uuids(self, sr_uuid=None):
+    def vdi_uuids(self, sr_uuid=None) -> List[str]:
         output = self._disk_list()
         if output == '':
             return []
@@ -77,6 +88,9 @@ def vdi_uuids(self, sr_uuid=None):
 
     def destroy_vdi(self, vdi_uuid: str) -> None:
         self.host.xe('vdi-destroy', {'uuid': vdi_uuid})
+        for vdi in self.vdis:
+            if vdi.uuid == vdi_uuid:
+                self.vdis.remove(vdi)
 
     def all_vdis_on_host(self, host):
         for vdi_uuid in self.vdi_uuids():

lib/host.py

@@ -67,6 +67,7 @@ def __init__(self, pool: 'lib.pool.Pool', hostname_or_ip):
         self.uuid = self.inventory['INSTALLATION_UUID']
         self.xcp_version = version.parse(self.inventory['PRODUCT_VERSION'])
         self.xcp_version_short = f"{self.xcp_version.major}.{self.xcp_version.minor}"
+        self._dom0: Optional[VM] = None
 
     def __str__(self):
         return self.hostname_or_ip
@@ -711,3 +712,20 @@ def enable_hsts_header(self):
     def disable_hsts_header(self):
         self.ssh(['rm', '-f', f'{XAPI_CONF_DIR}/00-XCP-ng-tests-enable-hsts-header.conf'])
         self.restart_toolstack(verify=True)
+
+    def get_dom0_uuid(self):
+        return self.inventory["CONTROL_DOMAIN_UUID"]
+
+    def get_dom0_VM(self) -> VM:
+        if not self._dom0:
+            self._dom0 = VM(self.get_dom0_uuid(), self)
+        return self._dom0
+
+    def get_sr_from_vdi_uuid(self, vdi_uuid: str) -> Optional[SR]:
+        sr_uuid = self.xe("vdi-param-get", {
+            "param-name": "sr-uuid",
+            "uuid": vdi_uuid,
+        })
+        if not sr_uuid:
+            return None
+        return SR(sr_uuid, self.pool)

lib/sr.py

@@ -1,11 +1,18 @@
 import logging
 import time
-from typing import Optional
 
 import lib.commands as commands
-from lib.common import prefix_object_name, safe_split, strtobool, wait_for, wait_for_not
+from lib.common import (
+    prefix_object_name,
+    safe_split,
+    strtobool,
+    wait_for,
+    wait_for_not,
+)
 from lib.vdi import VDI
 
+from typing import Optional
+
 class SR:
     def __init__(self, uuid, pool):
         self.uuid = uuid

lib/vm.py

@@ -19,6 +19,7 @@
 )
 from lib.snapshot import Snapshot
 from lib.vbd import VBD
+from lib.vdi import VDI
 from lib.vif import VIF
 
 from typing import TYPE_CHECKING, List, Literal, Optional, Union, overload
@@ -287,6 +288,30 @@ def checkpoint(self) -> Snapshot:
                                                        'new-name-label': 'Checkpoint of %s' % self.uuid}),
                         self.host)
 
+    def connect_vdi(self, vdi: VDI, device: str = "autodetect") -> str:
+        logging.info(f">> Plugging VDI {vdi.uuid} on VM {self.uuid}")
+        vbd_uuid = self.host.xe("vbd-create", {
+            "vdi-uuid": vdi.uuid,
+            "vm-uuid": self.uuid,
+            "device": device,
+        })
+        self.host.xe("vbd-plug", {"uuid": vbd_uuid})
+
+        self.vdis.append(vdi)
+
+        return vbd_uuid
+
+    def disconnect_vdi(self, vdi: VDI):
+        logging.info(f"<< Unplugging VDI {vdi.uuid} from VM {self.uuid}")
+        assert vdi in self.vdis, "VDI {vdi.uuid} not in VM {self.uuid} VDI list"
+        vbd_uuid = self.host.xe("vbd-list", {
+            "vdi-uuid": vdi.uuid,
+            "vm-uuid": self.uuid
+        }, minimal=True)
+        self.host.xe("vbd-unplug", {"uuid": vbd_uuid})
+        self.host.xe("vbd-destroy", {"uuid": vbd_uuid})
+        self.vdis.remove(vdi)
+
     def vifs(self):
         _vifs = []
         for vif_uuid in safe_split(self.host.xe('vif-list', {'vm-uuid': self.uuid}, minimal=True)):