lib: Add some types

.. for functions that are going to be used by
test_plugin_nfs_on_on_slave().

Signed-off-by: Anthony PERARD <[email protected]>

Author: tperard

lib/basevm.py

@@ -1,6 +1,6 @@
 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
@@ -59,7 +59,7 @@ def name(self) -> str:
     def _disk_list(self):
         raise NotImplementedError()
 
-    def vdi_uuids(self, sr_uuid=None):
+    def vdi_uuids(self, sr_uuid: Optional[str] = None) -> List[str]:
         output = self._disk_list()
         if output == '':
             return []
@@ -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

@@ -661,8 +661,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, Union[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

@@ -117,7 +117,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

@@ -305,10 +305,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)