WIP common: introduce type aliases for HostAddress and DiskDevName

Will help readability and improve typechecking.

FIXME: start to use them

Author: ydirson

lib/commands.py

@@ -5,6 +5,7 @@
 from typing import List, Literal, overload, Union
 
 
+from lib.common import HostAddress
 import lib.config as config
 from lib.netutil import wrap_ip
 
@@ -139,27 +140,32 @@ def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warning
 # This function is kept short for shorter pytest traces upon SSH failures, which are common,
 # as pytest prints the whole function definition that raised the SSHCommandFailed exception
 @overload
-def ssh(hostname_or_ip: str, cmd: Union[str, List[str]], *, check: bool = True, simple_output: Literal[True] = True,
+def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
+        simple_output: Literal[True] = True,
         suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
         decode: Literal[True] = True, options: List[str] = []) -> str:
     ...
 @overload
-def ssh(hostname_or_ip: str, cmd: Union[str, List[str]], *, check: bool = True, simple_output: Literal[True] = True,
+def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
+        simple_output: Literal[True] = True,
         suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
         decode: Literal[False], options: List[str] = []) -> bytes:
     ...
 @overload
-def ssh(hostname_or_ip: str, cmd: Union[str, List[str]], *, check: bool = True, simple_output: Literal[False],
+def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
+        simple_output: Literal[False],
         suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
         decode: bool = True, options: List[str] = []) -> SSHResult:
     ...
 @overload
-def ssh(hostname_or_ip: str, cmd: Union[str, List[str]], *, check: bool = True, simple_output: Literal[False],
+def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
+        simple_output: Literal[False],
         suppress_fingerprint_warnings: bool = True, background: Literal[True],
         decode: bool = True, options: List[str] = []) -> None:
     ...
 @overload
-def ssh(hostname_or_ip: str, cmd: Union[str, List[str]], *, check=True, simple_output: bool = True,
+def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check=True,
+        simple_output: bool = True,
         suppress_fingerprint_warnings=True, background: bool = False,
         decode: bool = True, options: List[str] = []) \
         -> Union[str, bytes, SSHResult, None]:

lib/common.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import getpass
 import inspect
 import itertools
@@ -16,9 +18,13 @@
 import lib.commands as commands
 if TYPE_CHECKING:
     import lib.host
+    from lib.typing import TypeAlias
 
 T = TypeVar("T")
 
+HostAddress: TypeAlias = str
+DiskDevName: TypeAlias = str
+
 class PackageManagerEnum(Enum):
     UNKNOWN = 1
     RPM = 2

lib/host.py

@@ -15,6 +15,7 @@
 if TYPE_CHECKING:
     import lib.pool
 
+from lib.common import DiskDevName
 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
@@ -534,13 +535,13 @@ def management_pif(self):
         uuid = self.xe('pif-list', {'management': True, 'host-uuid': self.uuid}, minimal=True)
         return pif.PIF(uuid, self)
 
-    def disks(self):
+    def disks(self) -> list[DiskDevName]:
         """ List of SCSI disks, e.g ['sda', 'sdb', 'nvme0n1']. """
         disks = self.ssh(['lsblk', '-nd', '-I', '8,259', '--output', 'NAME']).splitlines()
         disks.sort()
         return disks
 
-    def disk_is_available(self, disk: str) -> bool:
+    def disk_is_available(self, disk: DiskDevName) -> bool:
         """
         Check if a disk is unmounted and appears available for use.
 
@@ -552,7 +553,7 @@ def disk_is_available(self, disk: str) -> bool:
         """
         return len(self.ssh(['lsblk', '-n', '-o', 'MOUNTPOINT', '/dev/' + disk]).strip()) == 0
 
-    def available_disks(self, blocksize=512):
+    def available_disks(self, blocksize: int = 512) -> list[DiskDevName]:
         """
         Return a list of available disks for formatting, creating SRs or such.
 

lib/pool.py

@@ -6,6 +6,7 @@
 from packaging import version
 
 import lib.commands as commands
+from lib.common import HostAddress
 from lib.common import _param_get, _param_set, safe_split, wait_for, wait_for_not
 from lib.host import Host
 from lib.sr import SR
@@ -14,7 +15,7 @@
 class Pool:
     xe_prefix = "pool"
 
-    def __init__(self, master_hostname_or_ip: str) -> None:
+    def __init__(self, master_hostname_or_ip: HostAddress) -> None:
         master = Host(self, master_hostname_or_ip)
         assert master.is_master(), f"Host {master_hostname_or_ip} is not a master host. Aborting."
         self.master = master

lib/typing.py

@@ -1,6 +1,11 @@
 import sys
 from typing import TypedDict
 
+if sys.version_info >= (3, 10):
+    from typing import TypeAlias
+else:
+    from typing_extensions import TypeAlias
+
 if sys.version_info >= (3, 11):
     from typing import NotRequired
 else: