refactor: use the more modern | notation in type hints

Signed-off-by: Gaëtan Lehmann <gaetan.lehmann@vates.tech>

Author: glehmann

lib/basevm.py

@@ -2,7 +2,7 @@
 
 import logging
 
-from typing import TYPE_CHECKING, List, Literal, Optional, overload
+from typing import TYPE_CHECKING, List, Literal, overload
 
 if TYPE_CHECKING:
     from lib.host import Host
@@ -22,17 +22,17 @@ def __init__(self, uuid: str, host: Host):
         self.host = host
 
     @overload
-    def param_get(self, param_name: str, key: Optional[str] = ...,
+    def param_get(self, param_name: str, key: str | None = ...,
                   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: str | None = ...,
+                  accept_unknown_key: Literal[True] = ...) -> str | None:
         ...
 
-    def param_get(self, param_name: str, key: Optional[str] = None,
-                  accept_unknown_key: bool = False) -> Optional[str]:
+    def param_get(self, param_name: str, key: str | None = None,
+                  accept_unknown_key: bool = False) -> str | None:
         return _param_get(self.host, self.xe_prefix, self.uuid,
                           param_name, key, accept_unknown_key)
 
@@ -61,7 +61,7 @@ def name(self) -> str:
     def _disk_list(self) -> str:
         raise NotImplementedError()
 
-    def vdi_uuids(self, sr_uuid: Optional[str] = None) -> List[str]:
+    def vdi_uuids(self, sr_uuid: str | None = None) -> List[str]:
         output = self._disk_list()
         if output == '':
             return []

lib/commands.py

@@ -9,7 +9,7 @@
 import lib.config as config
 from lib.netutil import wrap_ip
 
-from typing import TYPE_CHECKING, List, Literal, Union, overload
+from typing import TYPE_CHECKING, List, Literal, overload
 
 if TYPE_CHECKING:
     from lib.common import HostAddress
@@ -137,7 +137,7 @@ def _ssh(
     if res.returncode == 255:
         return SSHCommandFailed(255, "SSH Error: %s" % output_for_errors, cmd)
 
-    output: Union[bytes, str] = res.stdout
+    output: str | bytes = res.stdout
     if banner_res:
         if banner_res.returncode == 255:
             return SSHCommandFailed(255, "SSH Error: %s" % banner_res.stdout.decode(errors='replace'), cmd)
@@ -186,7 +186,7 @@ def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True,
         simple_output: bool = True,
         suppress_fingerprint_warnings: bool = True, background: bool = False,
         decode: bool = True, options: List[str] = [], multiplexing: bool = True) \
-        -> Union[str, bytes, SSHResult, None]:
+        -> str | bytes | SSHResult | None:
     ...
 def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True, simple_output: bool = True,
         suppress_fingerprint_warnings: bool = True,

lib/common.py

@@ -25,10 +25,8 @@
     Any,
     Callable,
     Literal,
-    Optional,
     TypeAlias,
     TypeVar,
-    Union,
     cast,
     overload,
 )
@@ -134,7 +132,7 @@ def ensure_type(tp: Any, v: object) -> Any:
     """
     return _get_type_adapter(tp).validate_python(v)
 
-def callable_marker(value: Union[T, Callable[..., T]], request: pytest.FixtureRequest) -> T:
+def callable_marker(value: T | Callable[..., T], request: pytest.FixtureRequest) -> T:
     """
     Process value optionally generated by fixture-dependent callable.
 
@@ -304,22 +302,22 @@ def randid(length: int = 6) -> str:
     return ''.join(random.choices(characters, k=length))
 
 @overload
-def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
+def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: str | None = ...,
                accept_unknown_key: Literal[False] = ...) -> str:
     ...
 
 @overload
-def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
-               accept_unknown_key: Literal[True] = ...) -> Optional[str]:
+def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: str | None = ...,
+               accept_unknown_key: Literal[True] = ...) -> str | None:
     ...
 
 @overload
-def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
-               accept_unknown_key: bool = ...) -> Optional[str]:
+def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: str | None = ...,
+               accept_unknown_key: bool = ...) -> str | None:
     ...
 
-def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = None,
-               accept_unknown_key: bool = False) -> Optional[str]:
+def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: str | None = None,
+               accept_unknown_key: bool = False) -> str | None:
     """ Common implementation for param_get. """
     import lib.commands as commands
     args: dict[str, str | bool | dict[str, str]] = {'uuid': uuid, 'param-name': param_name}

lib/efi.py

@@ -21,7 +21,7 @@
 
 import lib.commands as commands
 
-from typing import Any, Iterable, Literal, Optional, Self, Union, cast
+from typing import Any, Iterable, Literal, Self, cast
 
 class _EfiGlobalTempdir:
     _instance = None
@@ -230,7 +230,7 @@ def certs_to_sig_db(certs: list[str] | str) -> bytes:
 
 
 def sign_efi_sig_db(
-    sig_db: bytes, var: str, key: str, cert: str, time: Optional[datetime] = None, guid: Optional[GUID] = None
+    sig_db: bytes, var: str, key: str, cert: str, time: datetime | None = None, guid: GUID | None = None
 ) -> bytes:
     """Return a pkcs7 SignedData from a UEFI signature database."""
     global p7_out
@@ -363,7 +363,7 @@ def pesign(key: str, cert: str, name: str, image: str) -> str:
 
 
 class Certificate:
-    def __init__(self, pub: str, key: Optional[str]):
+    def __init__(self, pub: str, key: str | None) -> None:
         self.pub = pub
         self.key = key
 
@@ -397,15 +397,15 @@ def copy(self) -> Certificate:
 
 
 class EFIAuth:
-    _auth_data: Optional[bytes]
+    _auth_data: bytes | None
     name: Literal["PK", "KEK", "db", "dbx"]
 
     def __init__(
         self,
         name: Literal["PK", "KEK", "db", "dbx"],
-        owner_cert: Optional[Certificate] = None,
-        other_certs: Optional[Iterable[Union[Certificate, str]]] = None,
-    ):
+        owner_cert: Certificate | None = None,
+        other_certs: Iterable[Certificate | str] | None = None,
+    ) -> None:
         assert name in SECURE_BOOT_VARIABLES
         assert owner_cert is None or owner_cert.key is not None, "owner cert must have private key"
         self.name = name
@@ -418,7 +418,7 @@ def __init__(
 
     @classmethod
     def self_signed(
-        cls, name: Literal["PK", "KEK", "db", "dbx"], other_certs: Optional[Iterable[Union[Certificate, str]]] = None
+        cls, name: Literal["PK", "KEK", "db", "dbx"], other_certs: Iterable[Certificate | str] | None = None
     ) -> Self:
         return cls(name, owner_cert=Certificate.self_signed(name + " Owner"), other_certs=other_certs)
 

lib/host.py

@@ -31,7 +31,7 @@
 from lib.vm import VM
 from lib.xo import xo_cli, xo_object_exists
 
-from typing import TYPE_CHECKING, Literal, Optional, TypedDict, Union, cast, overload
+from typing import TYPE_CHECKING, Literal, TypedDict, cast, overload
 
 if TYPE_CHECKING:
     from lib.pool import Pool
@@ -52,7 +52,7 @@ def host_data(hostname_or_ip: str) -> dict[str, str]:
 
 class Host:
     xe_prefix = "host"
-    pool: Pool
+    pool: "Pool"
 
     # Data extraction is automatic, no conversion from str is done.
     BlockDeviceInfo = TypedDict('BlockDeviceInfo', {"name": str,
@@ -69,7 +69,7 @@ class Host:
     def __init__(self, pool: Pool, hostname_or_ip: str):
         self.pool = pool
         self.hostname_or_ip = hostname_or_ip
-        self.xo_srv_id: Optional[str] = None
+        self.xo_srv_id: str | None = None
 
         h_data = host_data(self.hostname_or_ip)
         self.user = h_data['user']
@@ -82,7 +82,7 @@ def __init__(self, pool: Pool, hostname_or_ip: str):
         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
+        self._dom0: VM | None = None
 
         self.rescan_block_devices_info()
 
@@ -117,7 +117,7 @@ def ssh(self, cmd: str, *, check: bool = True, simple_output: bool = True,
     def ssh(self, cmd: str, *, check: bool = True, simple_output: bool = True,
             suppress_fingerprint_warnings: bool = True, background: bool = False, decode: bool = True,
             multiplexing: bool = True) \
-            -> Union[str, bytes, commands.SSHResult, None]:
+            -> str | bytes | commands.SSHResult | None:
         ...
 
     def ssh(self, cmd: str, *, check: bool = True, simple_output: bool = True,
@@ -150,7 +150,7 @@ def xe(self, action: str, args: dict[str, str | bool | dict[str, str]] = {}, *,
 
     def xe(self, action: str, args: dict[str, str | bool | dict[str, str]] = {}, *, check: bool = True,
            simple_output: bool = True, minimal: bool = False, force: bool = False) \
-            -> Union[str, commands.SSHResult]:
+            -> str | commands.SSHResult:
         maybe_param_minimal = '--minimal' if minimal else ''
         maybe_param_force = '--force' if force else ''
 
@@ -176,16 +176,16 @@ def stringify(key: str, value: str | bool | dict[str, str]) -> str:
         return result
 
     @overload
-    def param_get(self, param_name: str, key: Optional[str] = ...,
+    def param_get(self, param_name: str, key: str | None = ...,
                   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: str | None = ...,
+                  accept_unknown_key: Literal[True] = ...) -> str | None:
         ...
 
-    def param_get(self, param_name: str, key: Optional[str] = None, accept_unknown_key: bool = False) -> Optional[str]:
+    def param_get(self, param_name: str, key: str | None = None, accept_unknown_key: bool = False) -> str | None:
         return _param_get(self, self.xe_prefix, self.uuid,
                           param_name, key, accept_unknown_key)
 
@@ -358,8 +358,8 @@ def cached_vm(self, uri: str, sr_uuid: str) -> VM | None:
         logging.info("Could not find a VM in cache for %r", uri)
         return None
 
-    def import_vm(self, uri: str, sr_uuid: str | None = None, use_cache: bool = False) -> "VM":
-        vm = None
+    def import_vm(self, uri: str, sr_uuid: str | None = None, use_cache: bool = False) -> VM:
+        vm: VM | None = None
         if use_cache:
             assert sr_uuid is not None
             if '://' in uri and uri.startswith("clone"):
@@ -772,7 +772,7 @@ def get_dom0_vm(self) -> VM:
             self._dom0 = VM(self.get_dom0_uuid(), self)
         return self._dom0
 
-    def get_sr_from_vdi_uuid(self, vdi_uuid: str) -> Optional[SR]:
+    def get_sr_from_vdi_uuid(self, vdi_uuid: str) -> SR | None:
         sr_uuid = self.xe("vdi-param-get", {
             "param-name": "sr-uuid",
             "uuid": vdi_uuid,

lib/pool.py

@@ -12,7 +12,7 @@
 from lib.host import Host
 from lib.sr import SR
 
-from typing import Any, Callable, Iterable, Optional
+from typing import Any, Callable, Iterable
 
 class Pool:
     xe_prefix = "pool"
@@ -111,13 +111,13 @@ def get_host_by_uuid(self, host_uuid: str) -> Host:
                 return host
         raise Exception(f"Host with uuid {host_uuid} not found in pool.")
 
-    def first_host_that_isnt(self, host: Host) -> Optional[Host]:
+    def first_host_that_isnt(self, host: Host) -> Host | None:
         for h in self.hosts:
             if h != host:
                 return h
         return None
 
-    def first_shared_sr(self) -> Optional[SR]:
+    def first_shared_sr(self) -> SR | None:
         uuids = safe_split(self.master.xe('sr-list', {'shared': True, 'content-type': 'user'}, minimal=True))
         if len(uuids) > 0:
             return SR(uuids[0], self)

lib/sr.py

@@ -15,7 +15,7 @@
 )
 from lib.vdi import VDI, ImageFormat
 
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from lib.host import Host
@@ -25,9 +25,9 @@ class SR:
     def __init__(self, uuid: str, pool: Pool):
         self.uuid = uuid
         self.pool = pool
-        self._is_shared: Optional[bool] = None # cached value for is_shared()
-        self._main_host: Optional[Host] = None # cached value for main_host()
-        self._type: Optional[str] = None # cache value for get_type()
+        self._is_shared: bool | None = None # cached value for is_shared()
+        self._main_host: Host | None = None # cached value for main_host()
+        self._type: str | None = None # cache value for get_type()
 
     def pbd_uuids(self) -> list[str]:
         return safe_split(self.pool.master.xe('pbd-list', {'sr-uuid': self.uuid}, minimal=True))

lib/vdi.py

@@ -13,7 +13,7 @@
     wait_for_not,
 )
 
-from typing import TYPE_CHECKING, Callable, Literal, Optional, TypeVar, overload
+from typing import TYPE_CHECKING, Callable, Literal, TypeVar, overload
 
 if TYPE_CHECKING:
     from lib.host import Host
@@ -74,7 +74,7 @@ def resize(self, new_size: int) -> None:
     def __str__(self) -> str:
         return f"VDI {self.uuid} on SR {self.sr.uuid}"
 
-    def get_parent(self) -> Optional[str]:
+    def get_parent(self) -> str | None:
         return self.param_get("sm-config", key="vhd-parent", accept_unknown_key=True)
 
     def get_image_format(self) -> ImageFormat | None:
@@ -84,17 +84,17 @@ def get_image_format(self) -> ImageFormat | None:
         return ensure_type(ImageFormat, v)
 
     @overload
-    def param_get(self, param_name: str, key: Optional[str] = ...,
+    def param_get(self, param_name: str, key: str | None = ...,
                   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: str | None = ...,
+                  accept_unknown_key: Literal[True] = ...) -> str | None:
         ...
 
-    def param_get(self, param_name: str, key: Optional[str] = None,
-                  accept_unknown_key: bool = False) -> Optional[str]:
+    def param_get(self, param_name: str, key: str | None = None,
+                  accept_unknown_key: bool = False) -> str | None:
         return _param_get(self.sr.pool.master, self.xe_prefix, self.uuid,
                           param_name, key, accept_unknown_key)
 
@@ -116,7 +116,7 @@ def param_remove(self, param_name: str, key: str, accept_unknown_key: bool = Fal
 
     def wait_for_coalesce(self, fn: Callable[[], R] | None = None) -> R | None:
         previous_parent = self.get_parent()
-        ret = None
+        ret: R | None = None
         if fn is not None:
             ret = fn()
         # It is necessary to wait a long time because the GC can be paused for more than 5 minutes.

lib/vm.py

@@ -27,16 +27,16 @@
 from lib.vdi import VDI
 from lib.vif import VIF
 
-from typing import TYPE_CHECKING, Iterable, List, Literal, Optional, cast, overload
+from typing import TYPE_CHECKING, Iterable, List, Literal, cast, overload
 
 if TYPE_CHECKING:
     from lib.host import Host
 
 class VM(BaseVM):
-    def __init__(self, uuid: str, host: Host):
+    def __init__(self, uuid: str, host: Host) -> None:
         super().__init__(uuid, host)
-        self.ip: Optional[str] = None
-        self.previous_host: Optional[Host] = None # previous host when migrated or being migrated
+        self.ip: str | None = None
+        self.previous_host: Host | None = None # previous host when migrated or being migrated
         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'
         self.create_vdis_list()