Merge pull request #385 from semarie/push-xzykrvyyrtyq

Author: glehmann

conftest.py

@@ -187,7 +187,7 @@ def setup_host(hostname_or_ip, *, config=None):
             nested_list.append(host_vm)
 
             vif = host_vm.vifs()[0]
-            mac_address = vif.param_get('MAC')
+            mac_address = vif.mac_address()
             logging.info("Nested host has MAC %s", mac_address)
 
             host_vm.start()

lib/basevm.py

@@ -1,9 +1,11 @@
+from __future__ import annotations
+
 import logging
 
 from typing import TYPE_CHECKING, Any, List, Literal, Optional, overload
 
 if TYPE_CHECKING:
-    import lib.host
+    from lib.host import Host
 
 from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set
 from lib.sr import SR
@@ -14,7 +16,7 @@ class BaseVM:
     xe_prefix = "vm"
     uuid: str
 
-    def __init__(self, uuid: str, host: 'lib.host.Host'):
+    def __init__(self, uuid: str, host: Host):
         logging.info("New %s: %s", type(self).__name__, uuid)
         self.uuid = uuid
         self.host = host
@@ -75,14 +77,14 @@ def vdi_uuids(self, sr_uuid: Optional[str] = None) -> List[str]:
                 vdis_on_sr.append(vdi)
         return vdis_on_sr
 
-    def all_vdis_on_host(self, host):
+    def all_vdis_on_host(self, host: Host) -> bool:
         for vdi_uuid in self.vdi_uuids():
             sr = SR(self.host.pool.get_vdi_sr_uuid(vdi_uuid), self.host.pool)
             if not sr.attached_to_host(host):
                 return False
         return True
 
-    def all_vdis_on_sr(self, sr) -> bool:
+    def all_vdis_on_sr(self, sr: 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) -> SR:

lib/host.py

@@ -10,7 +10,6 @@
 from packaging import version
 
 import lib.commands as commands
-import lib.pif as pif
 
 from typing import TYPE_CHECKING, Dict, List, Literal, Mapping, Optional, TypedDict, Union, overload
 
@@ -33,6 +32,7 @@
     wait_for_not,
 )
 from lib.netutil import wrap_ip
+from lib.pif import PIF
 from lib.sr import SR
 from lib.vdi import VDI
 from lib.vm import VM
@@ -313,7 +313,7 @@ def xo_server_reconnect(self):
     def vm_cache_key(uri):
         return f"[Cache for {strip_suffix(uri, '.xva')}]"
 
-    def cached_vm(self, uri, sr_uuid):
+    def cached_vm(self, uri, sr_uuid) -> Optional[VM]:
         assert sr_uuid, "A SR UUID is necessary to use import cache"
         cache_key = self.vm_cache_key(uri)
         # Look for an existing cache VM
@@ -328,8 +328,9 @@ def cached_vm(self, uri, sr_uuid):
                 logging.info(f"Reusing cached VM {vm.uuid} for {uri}")
                 return vm
         logging.info("Could not find a VM in cache for %r", uri)
+        return None
 
-    def import_vm(self, uri, sr_uuid=None, use_cache=False):
+    def import_vm(self, uri, sr_uuid=None, use_cache=False) -> VM:
         vm = None
         if use_cache:
             if '://' in uri and uri.startswith("clone"):
@@ -373,7 +374,7 @@ def import_vm(self, uri, sr_uuid=None, use_cache=False):
             vm.param_set('name-description', cache_key)
         return vm
 
-    def import_iso(self, uri, sr: SR):
+    def import_iso(self, uri, sr: SR) -> VDI:
         random_name = str(uuid.uuid4())
 
         vdi_uuid = self.xe(
@@ -404,7 +405,7 @@ def import_iso(self, uri, sr: SR):
 
         return VDI(vdi_uuid, sr=sr)
 
-    def vm_from_template(self, name, template):
+    def vm_from_template(self, name, template) -> VM:
         params = {
             "new-name-label": prefix_object_name(name),
             "template": template,
@@ -413,7 +414,7 @@ def vm_from_template(self, name, template):
         vm_uuid = self.xe('vm-install', params)
         return VM(vm_uuid, self)
 
-    def pool_has_vm(self, vm_uuid, vm_type='vm'):
+    def pool_has_vm(self, vm_uuid, vm_type='vm') -> bool:
         if vm_type == 'snapshot':
             return self.xe('snapshot-list', {'uuid': vm_uuid}, minimal=True) == vm_uuid
         else:
@@ -436,7 +437,7 @@ def is_enabled(self) -> bool:
             # If XAPI is not ready yet, or the host is down, this will throw. We return False in that case.
             return False
 
-    def has_updates(self):
+    def has_updates(self) -> bool:
         try:
             # yum check-update returns 100 if there are updates, 1 if there's an error, 0 if no updates
             self.ssh(['yum', 'check-update'])
@@ -504,14 +505,14 @@ def packages(self):
             self.ssh(['rpm', '-qa', '--qf', '%{NAME}-%{VERSION}-%{RELEASE}-%{ARCH}-%{EPOCH}\\\\n']).splitlines()
         )
 
-    def check_packages_available(self, packages):
+    def check_packages_available(self, packages) -> bool:
         """ Check if a given package list is available in the YUM repositories. """
         return len(self.ssh(['repoquery'] + packages).splitlines()) == len(packages)
 
     def get_available_package_versions(self, package):
         return self.ssh(['repoquery', '--show-duplicates', package]).splitlines()
 
-    def is_package_installed(self, package):
+    def is_package_installed(self, package) -> bool:
         return self.ssh_with_result(['rpm', '-q', package]).returncode == 0
 
     def yum_save_state(self):
@@ -563,12 +564,12 @@ def reboot(self, verify=False):
                      "Wait for ssh up on host", timeout_secs=10 * 60, retry_delay_secs=5)
             wait_for(self.is_enabled, "Wait for XAPI to be ready", timeout_secs=30 * 60)
 
-    def management_network(self):
+    def management_network(self) -> str:
         return self.xe('network-list', {'bridge': self.inventory['MANAGEMENT_INTERFACE']}, minimal=True)
 
-    def management_pif(self):
+    def management_pif(self) -> PIF:
         uuid = self.xe('pif-list', {'management': True, 'host-uuid': self.uuid}, minimal=True)
-        return pif.PIF(uuid, self)
+        return PIF(uuid, self)
 
     def rescan_block_devices_info(self) -> None:
         """
@@ -607,17 +608,17 @@ def disk_is_available(self, disk: DiskDevName) -> bool:
         """
         return len(self.ssh(['lsblk', '--noheadings', '-o', 'MOUNTPOINT', '/dev/' + disk]).strip()) == 0
 
-    def file_exists(self, filepath, regular_file=True):
+    def file_exists(self, filepath, regular_file=True) -> bool:
         option = '-f' if regular_file else '-e'
         return self.ssh_with_result(['test', option, filepath]).returncode == 0
 
-    def binary_exists(self, binary):
+    def binary_exists(self, binary) -> bool:
         return self.ssh_with_result(['which', binary]).returncode == 0
 
-    def is_symlink(self, filepath):
+    def is_symlink(self, filepath) -> bool:
         return self.ssh_with_result(['test', '-L', filepath]).returncode == 0
 
-    def sr_create(self, sr_type, label, device_config, shared=False, verify=False):
+    def sr_create(self, sr_type, label, device_config, shared=False, verify=False) -> SR:
         params = {
             'host-uuid': self.uuid,
             'type': sr_type,
@@ -637,10 +638,10 @@ def sr_create(self, sr_type, label, device_config, shared=False, verify=False):
             wait_for(sr.exists, "Wait for SR to exist")
         return sr
 
-    def is_master(self):
+    def is_master(self) -> bool:
         return self.ssh(['cat', '/etc/xensource/pool.conf']) == 'master'
 
-    def local_vm_srs(self):
+    def local_vm_srs(self) -> list[SR]:
         srs = []
         sr_uuids = safe_split(self.xe('pbd-list', {'host-uuid': self.uuid, 'params': 'sr-uuid'}, minimal=True))
         for sr_uuid in sr_uuids:
@@ -649,7 +650,7 @@ def local_vm_srs(self):
                 srs.append(sr)
         return srs
 
-    def main_sr_uuid(self):
+    def main_sr_uuid(self) -> str:
         """ Main SR is the default SR, the first local SR, or a specific SR depending on data.py's DEFAULT_SR. """
         try:
             from data import DEFAULT_SR
@@ -695,7 +696,7 @@ def call_plugin(self, plugin_name: str, function: str,
                 params['args:%s' % k] = v
         return self.xe('host-call-plugin', params)
 
-    def join_pool(self, pool):
+    def join_pool(self, pool: Pool):
         master = pool.master
         self.xe('pool-join', {
             'master-address': master.hostname_or_ip,

lib/pool.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 import os
 import traceback
@@ -95,19 +97,19 @@ def exec_on_hosts_on_error_continue(self, func, host_list=[]):
         if errors:
             raise Exception(f"One or more exceptions were raised in `exec_on_hosts_on_error_continue`: {errors}")
 
-    def hosts_uuids(self):
+    def hosts_uuids(self) -> list[str]:
         return safe_split(self.master.xe('host-list', {}, minimal=True))
 
-    def host_ip(self, host_uuid):
+    def host_ip(self, host_uuid) -> str:
         return self.master.xe('host-param-get', {'uuid': host_uuid, 'param-name': 'address'})
 
-    def get_host_by_uuid(self, host_uuid):
+    def get_host_by_uuid(self, host_uuid) -> Host:
         for host in self.hosts:
             if host.uuid == host_uuid:
                 return host
         raise Exception(f"Host with uuid {host_uuid} not found in pool.")
 
-    def first_host_that_isnt(self, host):
+    def first_host_that_isnt(self, host: Host) -> Optional[Host]:
         for h in self.hosts:
             if h != host:
                 return h
@@ -122,15 +124,15 @@ def first_shared_sr(self) -> Optional[SR]:
     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):
+    def get_iso_sr(self) -> SR:
         uuids = safe_split(self.master.xe('sr-list', {'type': 'iso',
                                                       'content-type': 'iso',
                                                       'is-tools-sr': False},
                                           minimal=True))
         assert len(uuids) == 1  # we may need to allow finer selection if this triggers
         return SR(uuids[0], self)
 
-    def push_iso(self, local_file, remote_filename=None):
+    def push_iso(self, local_file, remote_filename=None) -> str:
         iso_sr = self.get_iso_sr()
         mountpoint = f"/run/sr-mount/{iso_sr.uuid}"
         if remote_filename is None:
@@ -280,7 +282,7 @@ def install_custom_uefi_certs(self, auths: Iterable[EFIAuth]):
         finally:
             host.ssh(['rm', '-f'] + list(auths_dict.values()))
 
-    def eject_host(self, host):
+    def eject_host(self, host: Host):
         master = self.master
         master.xe('pool-eject', {'host-uuid': host.uuid, 'force': True})
         wait_for_not(lambda: host.uuid in self.hosts_uuids(), f"Wait for host {host} to be ejected of pool {master}.")

lib/sr.py

@@ -1,6 +1,14 @@
+from __future__ import annotations
+
 import logging
 import time
 
+from typing import TYPE_CHECKING, Optional
+
+if TYPE_CHECKING:
+    from lib.host import Host
+    from lib.pool import Pool
+
 import lib.commands as commands
 from lib.common import (
     GiB,
@@ -14,12 +22,12 @@
 from lib.vdi import VDI, ImageFormat
 
 class SR:
-    def __init__(self, uuid, pool):
+    def __init__(self, uuid, pool: Pool):
         self.uuid = uuid
         self.pool = pool
-        self._is_shared = None # cached value for is_shared()
-        self._main_host = None # cached value for main_host()
-        self._type = None # cache value for get_type()
+        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()
 
     def pbd_uuids(self):
         return safe_split(self.pool.master.xe('pbd-list', {'sr-uuid': self.uuid}, minimal=True))
@@ -143,7 +151,7 @@ def hosts_uuids(self):
     def attached_to_host(self, host):
         return host.uuid in self.hosts_uuids()
 
-    def main_host(self):
+    def main_host(self) -> Host:
         """ Returns the host in case of a local SR, the master host in case of a shared SR. """
         if self._main_host is None:
             if self.is_shared():
@@ -155,7 +163,7 @@ def main_host(self):
     def content_type(self):
         return self.pool.master.xe('sr-param-get', {'uuid': self.uuid, 'param-name': 'content-type'})
 
-    def is_shared(self):
+    def is_shared(self) -> bool:
         if self._is_shared is None:
             self._is_shared = strtobool(self.pool.master.xe('sr-param-get',
                                                             {'uuid': self.uuid, 'param-name': 'shared'}))

lib/vif.py

@@ -39,3 +39,8 @@ def move(self, network_uuid):
 
     def destroy(self):
         self.vm.host.xe('vif-destroy', {'uuid': self.uuid})
+
+    def mac_address(self) -> str:
+        mac_address = self.param_get('MAC')
+        assert mac_address is not None, "VIF must have a MAC address"
+        return mac_address