Adapt return types of xe and param_get

Simplify the return type of `xe`, remove bool being a implicit cast for
`xe`.
Adapt some call that were expecting a bool to use `strtobool`
explicitely instead.
Adapted `_param_get` to new types for xe.

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

Author: Nambrok

lib/basevm.py

@@ -1,6 +1,6 @@
 import logging
 
-from typing import Any, Optional, TYPE_CHECKING, Union
+from typing import Any, Literal, Optional, overload, TYPE_CHECKING
 
 import lib.commands as commands
 if TYPE_CHECKING:
@@ -13,14 +13,25 @@ class BaseVM:
     """ Base class for VM and Snapshot. """
 
     xe_prefix = "vm"
+    uuid: str
 
-    def __init__(self, uuid, host: 'lib.host.Host'):
+    def __init__(self, uuid: str, host: 'lib.host.Host'):
         logging.info("New %s: %s", type(self).__name__, uuid)
         self.uuid = uuid
         self.host = host
 
+    @overload
+    def param_get(self, param_name: str, key: Optional[str] = ...,
+                  accept_unknown_key: Literal[False] = ...) -> str:
+        ...
+
+    @overload
+    def param_get(self, param_name: str, key: Optional[str] = ...,
+                  accept_unknown_key: Literal[True] = ...) -> Optional[str]:
+        ...
+
     def param_get(self, param_name: str, key: Optional[str] = None,
-                  accept_unknown_key: bool = False) -> Union[str, bool, None]:
+                  accept_unknown_key: bool = False) -> Optional[str]:
         return _param_get(self.host, self.xe_prefix, self.uuid,
                           param_name, key, accept_unknown_key)
 

lib/common.py

@@ -5,7 +5,7 @@
 import time
 import traceback
 from enum import Enum
-from typing import TYPE_CHECKING, Dict, Optional, Union
+from typing import Dict, Literal, Optional, overload, TYPE_CHECKING, Union
 from uuid import UUID
 
 import lib.commands as commands
@@ -63,7 +63,7 @@ def is_uuid(maybe_uuid):
     except ValueError:
         return False
 
-def to_xapi_bool(b):
+def to_xapi_bool(b: bool):
     return 'true' if b else 'false'
 
 def parse_xe_dict(xe_dict):
@@ -160,8 +160,23 @@ def strtobool(str):
         return False
     raise ValueError("invalid truth value '{}'".format(str))
 
+@overload
+def _param_get(host: 'lib.host.Host', xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
+               accept_unknown_key: Literal[False] = ...) -> str:
+    ...
+
+@overload
+def _param_get(host: 'lib.host.Host', xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
+               accept_unknown_key: Literal[True] = ...) -> Optional[str]:
+    ...
+
+@overload
+def _param_get(host: 'lib.host.Host', xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
+               accept_unknown_key: bool = ...) -> Optional[str]:
+    ...
+
 def _param_get(host: 'lib.host.Host', xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = None,
-               accept_unknown_key=False) -> Union[str, bool, None]:
+               accept_unknown_key: bool = False) -> Optional[str]:
     """ Common implementation for param_get. """
     args: Dict[str, Union[str, bool]] = {'uuid': uuid, 'param-name': param_name}
     if key is not None:

lib/host.py

@@ -8,14 +8,14 @@
 import uuid
 
 from packaging import version
-from typing import TYPE_CHECKING, Dict, List, Literal, Optional, Union, overload
+from typing import Dict, List, Literal, Optional, overload, TYPE_CHECKING, Union
 
 import lib.commands as commands
 import lib.pif as pif
 if TYPE_CHECKING:
     import lib.pool
 
-from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set
+from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set, strtobool
 from lib.common import safe_split, strip_suffix, to_xapi_bool, wait_for, wait_for_not
 from lib.common import prefix_object_name
 from lib.netutil import wrap_ip
@@ -106,17 +106,17 @@ def scp(self, src, dest, check=True, suppress_fingerprint_warnings=True, local_d
         )
 
     @overload
-    def xe(self, action: str, args: Dict[str, Union[str, bool]] = {}, *, check: bool = True,
-           simple_output: Literal[True] = True, minimal: bool = False, force: bool = False) -> Union[bool, str]:
+    def xe(self, action: str, args: Dict[str, Union[str, bool]] = {}, *, check: bool = ...,
+           simple_output: Literal[True] = ..., minimal: bool = ..., force: bool = ...) -> str:
         ...
 
     @overload
-    def xe(self, action: str, args: Dict[str, Union[str, bool]] = {}, *, check: bool = True,
-           simple_output: Literal[False], minimal: bool = False, force: bool = False) -> commands.SSHResult:
+    def xe(self, action: str, args: Dict[str, Union[str, bool]] = {}, *, check: bool = ...,
+           simple_output: Literal[False], minimal: bool = ..., force: bool = ...) -> commands.SSHResult:
         ...
 
     def xe(self, action, args={}, *, check=True, simple_output=True, minimal=False, force=False) \
-            -> Union[bool, str, commands.SSHResult]:
+            -> Union[str, commands.SSHResult]:
         maybe_param_minimal = ['--minimal'] if minimal else []
         maybe_param_force = ['--force'] if force else []
 
@@ -139,13 +139,19 @@ def stringify(key, value):
         )
         assert isinstance(result, (str, commands.SSHResult))
 
-        if result == 'true':
-            return True
-        if result == 'false':
-            return False
         return result
 
-    def param_get(self, param_name, key=None, accept_unknown_key=False):
+    @overload
+    def param_get(self, param_name: str, key: Optional[str] = ...,
+                  accept_unknown_key: Literal[False] = ...) -> str:
+        ...
+
+    @overload
+    def param_get(self, param_name: str, key: Optional[str] = ...,
+                  accept_unknown_key: Literal[True] = ...) -> Optional[str]:
+        ...
+
+    def param_get(self, param_name: str, key: Optional[str] = None, accept_unknown_key: bool = False) -> Optional[str]:
         return _param_get(self, self.xe_prefix, self.uuid,
                           param_name, key, accept_unknown_key)
 
@@ -330,7 +336,7 @@ def import_iso(self, uri, sr: SR):
 
         download_path = None
         try:
-            params = {'uuid': vdi_uuid}
+            params: Dict[str, Union[str, bool]] = {'uuid': vdi_uuid}
             if '://' in uri:
                 logging.info(f"Download ISO {uri}")
                 download_path = f'/tmp/{vdi_uuid}'
@@ -363,9 +369,9 @@ def restart_toolstack(self, verify=False):
         if verify:
             wait_for(self.is_enabled, "Wait for host enabled", timeout_secs=30 * 60)
 
-    def is_enabled(self):
+    def is_enabled(self) -> bool:
         try:
-            return self.param_get('enabled')
+            return strtobool(self.param_get('enabled'))
         except commands.SSHCommandFailed:
             # If XAPI is not ready yet, or the host is down, this will throw. We return False in that case.
             return False
@@ -621,9 +627,9 @@ def join_pool(self, pool):
             f"Wait for joining host {self} to appear in joined pool {master}."
         )
         pool.hosts.append(Host(pool, pool.host_ip(self.uuid)))
-        # Do not use `self.is_enabled` since it'd ask the XAPi of hostB1 before the join...
+        # Do not use `self.is_enabled` since it'd ask the XAPI of hostB1 before the join...
         wait_for(
-            lambda: master.xe('host-param-get', {'uuid': self.uuid, 'param-name': 'enabled'}),
+            lambda: strtobool(master.xe('host-param-get', {'uuid': self.uuid, 'param-name': 'enabled'})),
             f"Wait for pool {master} to see joined host {self} as enabled."
         )
         self.pool = pool

lib/pool.py

@@ -30,7 +30,7 @@ def __init__(self, master_hostname_or_ip: str) -> None:
             if host_uuid != self.hosts[0].uuid:
                 host = Host(self, self.host_ip(host_uuid))
                 self.hosts.append(host)
-        self.uuid = cast(str, self.master.xe('pool-list', minimal=True))
+        self.uuid = self.master.xe('pool-list', minimal=True)
         self.saved_uefi_certs: Optional[Dict[str, Any]] = None
 
     def param_get(self, param_name, key=None, accept_unknown_key=False):

lib/sr.py

@@ -3,7 +3,7 @@
 
 import lib.commands as commands
 
-from lib.common import prefix_object_name, safe_split, 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
 
 class SR:
@@ -41,8 +41,10 @@ def unplug_pbds(self, force=False):
     def all_pbds_attached(self):
         all_attached = True
         for pbd_uuid in self.pbd_uuids():
-            all_attached = all_attached and self.pool.master.xe('pbd-param-get', {'uuid': pbd_uuid,
-                                                                'param-name': 'currently-attached'})
+            all_attached = all_attached and strtobool(self.pool.master.xe('pbd-param-get',
+                                                                          {'uuid': pbd_uuid,
+                                                                           'param-name': 'currently-attached',
+                                                                           }))
         return all_attached
 
     def plug_pbd(self, pbd_uuid):
@@ -147,7 +149,8 @@ def content_type(self):
 
     def is_shared(self):
         if self._is_shared is None:
-            self._is_shared = self.pool.master.xe('sr-param-get', {'uuid': self.uuid, 'param-name': 'shared'})
+            self._is_shared = strtobool(self.pool.master.xe('sr-param-get',
+                                                            {'uuid': self.uuid, 'param-name': 'shared'}))
         return self._is_shared
 
     def create_vdi(self, name_label, virtual_size=64):

lib/vdi.py

@@ -1,6 +1,6 @@
 import logging
 
-from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set
+from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set, strtobool
 from typing import Literal, Optional, overload, TYPE_CHECKING
 if TYPE_CHECKING:
     from lib.host import Host
@@ -31,7 +31,7 @@ def __init__(self, uuid, *, host=None, sr=None):
         else:
             self.sr = sr
 
-    def name(self):
+    def name(self) -> str:
         return self.param_get('name-label')
 
     def destroy(self):
@@ -42,13 +42,24 @@ def clone(self):
         uuid = self.sr.pool.master.xe('vdi-clone', {'uuid': self.uuid})
         return VDI(uuid, sr=self.sr)
 
-    def readonly(self):
-        return self.param_get("read-only") == "true"
+    def readonly(self) -> bool:
+        return strtobool(self.param_get("read-only"))
 
     def __str__(self):
         return f"VDI {self.uuid} on SR {self.sr.uuid}"
 
-    def param_get(self, param_name, key=None, accept_unknown_key=False):
+    @overload
+    def param_get(self, param_name: str, key: Optional[str] = ...,
+                  accept_unknown_key: Literal[False] = ...) -> str:
+        ...
+
+    @overload
+    def param_get(self, param_name: str, key: Optional[str] = ...,
+                  accept_unknown_key: Literal[True] = ...) -> Optional[str]:
+        ...
+
+    def param_get(self, param_name: str, key: Optional[str] = None,
+                  accept_unknown_key: bool = False) -> Optional[str]:
         return _param_get(self.sr.pool.master, self.xe_prefix, self.uuid,
                           param_name, key, accept_unknown_key)
 

lib/vm.py

@@ -4,7 +4,7 @@
 import os
 import subprocess
 import tempfile
-from typing import List, Literal, Optional, Union, cast, overload
+from typing import Dict, List, Literal, Optional, overload, Union
 
 import lib.commands as commands
 import lib.efi as efi
@@ -22,7 +22,7 @@ def __init__(self, uuid, host):
         self.is_windows = self.param_get('platform', 'device_id', accept_unknown_key=True) == '0002'
         self.is_uefi = self.param_get('HVM-boot-params', 'firmware', accept_unknown_key=True) == 'uefi'
 
-    def power_state(self):
+    def power_state(self) -> str:
         return self.param_get('power-state')
 
     def is_running(self):
@@ -74,7 +74,7 @@ def reboot(self, force=False, verify=False):
         return ret
 
     def try_get_and_store_ip(self):
-        ip = cast(str, self.param_get('networks', '0/ip', accept_unknown_key=True))
+        ip = self.param_get('networks', '0/ip', accept_unknown_key=True)
 
         # An IP that starts with 169.254. is not a real routable IP.
         # VMs may return such an IP before they get an actual one from DHCP.
@@ -709,7 +709,7 @@ def are_windows_drivers_present(self):
 
     def are_windows_tools_working(self):
         assert self.is_windows
-        return self.is_windows_pv_device_installed() and self.param_get("PV-drivers-detected")
+        return self.is_windows_pv_device_installed() and strtobool(self.param_get("PV-drivers-detected"))
 
     def are_windows_tools_uninstalled(self):
         assert self.is_windows

tests/guest_tools/win/other_tools.py

@@ -2,14 +2,14 @@
 from pathlib import PureWindowsPath
 from typing import Any, Dict
 
-from lib.common import wait_for
+from lib.common import strtobool, wait_for
 from lib.vm import VM
 from . import WINDOWS_SHUTDOWN_COMMAND, enable_testsign, insert_cd_safe, wait_for_vm_running_and_ssh_up_without_tools
 
 
 def install_other_drivers(vm: VM, other_tools_iso_name: str, param: Dict[str, Any]):
     if param.get("vendor_device"):
-        assert not vm.param_get("has-vendor-device")
+        assert not strtobool(vm.param_get("has-vendor-device"))
         vm.param_set("has-vendor-device", True)
 
     vm.start()

tests/xapi/tls_verification/test_tls_verification.py

@@ -2,6 +2,7 @@
 import pytest
 
 from lib.commands import SSHCommandFailed
+from lib.common import strtobool
 
 # Requirements:
 # From --hosts parameter:
@@ -21,7 +22,7 @@
 def host_with_tls_verification_enabled(hostA1):
     for h in hostA1.pool.hosts:
         logging.info(f"Check that TLS verification is enabled on host {h}")
-        assert h.param_get("tls-verification-enabled"), f"TLS verification must be enabled on host {h}"
+        assert strtobool(h.param_get("tls-verification-enabled")), f"TLS verification must be enabled on host {h}"
         logging.info(f"Check that the host certificate exists on host {h}")
         cert_uuid = hostA1.xe('certificate-list', {'host': h.uuid, 'type': 'host_internal'}, minimal=True)
         assert len(cert_uuid) > 0, f"A host_internal certificate must exist on host {h}"