Implement storage nfs-on-slave test

data.py-dist

@@ -183,44 +183,44 @@ DEFAULT_SR = 'default'
 CACHE_IMPORTED_VM = False
 
 # Default NFS device config:
-NFS_DEVICE_CONFIG: dict[str, dict[str, str]] = {
+NFS_DEVICE_CONFIG: dict[str, str] = {
 #    'server': '10.0.0.2', # URL/Hostname of NFS server
 #    'serverpath': '/path/to/shared/mount' # Path to shared mountpoint
 }
 
 # Default NFS4+ only device config:
-NFS4_DEVICE_CONFIG: dict[str, dict[str, str]] = {
+NFS4_DEVICE_CONFIG: dict[str, str] = {
 #    'server': '10.0.0.2', # URL/Hostname of NFS server
 #    'serverpath': '/path_to_shared_mount' # Path to shared mountpoint
 #    'nfsversion': '4.1'
 }
 
 # Default NFS ISO device config:
-NFS_ISO_DEVICE_CONFIG: dict[str, dict[str, str]] = {
+NFS_ISO_DEVICE_CONFIG: dict[str, str] = {
 #    'location': '10.0.0.2:/path/to/shared/mount' # URL/Hostname of NFS server and path to shared mountpoint
 }
 
 # Default CIFS ISO device config:
-CIFS_ISO_DEVICE_CONFIG: dict[str, dict[str, str]] = {
+CIFS_ISO_DEVICE_CONFIG: dict[str, str] = {
 #    'location': r'\\10.0.0.2\<shared folder name>',
 #    'username': '<user>',
 #    'cifspassword': '<password>',
 #    'type': 'cifs',
 #    'vers': '<1.0> or <3.0>'
 }
 
-CEPHFS_DEVICE_CONFIG: dict[str, dict[str, str]] = {
+CEPHFS_DEVICE_CONFIG: dict[str, str] = {
 #    'server': '10.0.0.2',
 #    'serverpath': '/vms'
 }
 
-MOOSEFS_DEVICE_CONFIG: dict[str, dict[str, str]] = {
+MOOSEFS_DEVICE_CONFIG: dict[str, str] = {
 #    'masterhost': 'mfsmaster',
 #    'masterport': '9421',
 #    'rootpath': '/vms'
 }
 
-LVMOISCSI_DEVICE_CONFIG: dict[str, dict[str, str]] = {
+LVMOISCSI_DEVICE_CONFIG: dict[str, str] = {
 #    'target': '192.168.1.1',
 #    'port': '3260',
 #    'targetIQN': 'target.example',

lib/basevm.py

@@ -88,7 +88,7 @@ def all_vdis_on_host(self, host):
     def all_vdis_on_sr(self, sr) -> bool:
         return all(self.host.pool.get_vdi_sr_uuid(vdi_uuid) == sr.uuid for vdi_uuid in self.vdi_uuids())
 
-    def get_sr(self):
+    def get_sr(self) -> SR:
         # in this method we assume the SR of the first VDI is the VM SR
         vdis = self.vdi_uuids()
         assert len(vdis) > 0, "Don't ask for the SR of a VM without VDIs!"

lib/host.py

@@ -673,8 +673,13 @@ def main_sr_uuid(self):
     def hostname(self):
         return self.ssh(['hostname'])
 
-    def call_plugin(self, plugin_name, function, args=None):
-        params = {'host-uuid': self.uuid, 'plugin': plugin_name, 'fn': function}
+    def call_plugin(self, plugin_name: str, function: str,
+                    args: Optional[Dict[str, str]] = None) -> str:
+        params: Dict[str, str | bool] = {
+            'host-uuid': self.uuid,
+            'plugin': plugin_name,
+            'fn': function
+        }
         if args is not None:
             for k, v in args.items():
                 params['args:%s' % k] = v

lib/pool.py

@@ -118,7 +118,7 @@ def first_shared_sr(self) -> Optional[SR]:
             return SR(uuids[0], self)
         return None
 
-    def get_vdi_sr_uuid(self, vdi_uuid):
+    def get_vdi_sr_uuid(self, vdi_uuid: str) -> str:
         return self.master.xe('vdi-param-get', {'uuid': vdi_uuid, 'param-name': 'sr-uuid'})
 
     def get_iso_sr(self):

lib/vm.py

@@ -373,10 +373,10 @@ def create_vif(self, vif_num, *, network_uuid=None, network_name=None):
                                     'network-uuid': network_uuid,
                                     })
 
-    def is_running_on_host(self, host):
+    def is_running_on_host(self, host: Host) -> bool:
         return self.is_running() and self.param_get('resident-on') == host.uuid
 
-    def get_residence_host(self):
+    def get_residence_host(self) -> Host:
         assert self.is_running()
         host_uuid = self.param_get('resident-on')
         return self.host.pool.get_host_by_uuid(host_uuid)

tests/storage/nfs/test_nfs_sr.py

@@ -1,8 +1,17 @@
+from __future__ import annotations
+
 import pytest
 
+from lib.commands import SSHCommandFailed
 from lib.common import vm_image, wait_for
+from lib.vdi import VDI
 from tests.storage import vdi_is_open
 
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from lib.vm import VM
+
 # Requirements:
 # - one XCP-ng host >= 8.0 with an additional unused disk for the SR
 
@@ -30,6 +39,39 @@ def test_vdi_is_not_open(self, dispatch_nfs):
         vdi = dispatch_nfs
         assert not vdi_is_open(vdi)
 
+    @pytest.mark.small_vm
+    @pytest.mark.usefixtures('hostA2')
+    # Make sure this fixture is called before the parametrized one
+    @pytest.mark.usefixtures('vm_ref')
+    @pytest.mark.parametrize('dispatch_nfs', ['vm_on_nfs_sr', 'vm_on_nfs4_sr'], indirect=True)
+    def test_plugin_nfs_on_on_slave(self, dispatch_nfs: VM):
+        vm = dispatch_nfs
+        vm.start()
+        vm.wait_for_os_booted()
+        host = vm.get_residence_host()
+
+        vdi = vm.vdis[0]
+        image_format = vdi.get_image_format() or "vhd"
+
+        vdi_path = f"/run/sr-mount/{vdi.sr.uuid}/{vdi.uuid}.{image_format}"
+
+        # nfs-on-slave returns an error when the VDI is open on the host.
+        # Otherwise, it return "success", including in case "path" doesn't exist
+        with pytest.raises(SSHCommandFailed) as excinfo:
+            host.call_plugin("nfs-on-slave", "check", {"path": vdi_path})
+
+        # The output of the host plugin would have "stdout: NfsCheckException"
+        # and information about which process has the path open.
+        assert "NfsCheckException" in excinfo.value.stdout
+
+        for member in host.pool.hosts:
+            # skip the host where the VM is running
+            if member.uuid == host.uuid:
+                continue
+            member.call_plugin("nfs-on-slave", "check", {"path": vdi_path})
+
+        vm.shutdown(verify=True)
+
     @pytest.mark.small_vm # run with a small VM to test the features
     @pytest.mark.big_vm # and ideally with a big VM to test it scales
     # Make sure this fixture is called before the parametrized one