diff --git a/conftest.py b/conftest.py index 03ec108ea..f4cfed129 100644 --- a/conftest.py +++ b/conftest.py @@ -432,10 +432,9 @@ def vm_ref(request): if ref is None: # get default VM from test if there's one marker = request.node.get_closest_marker("default_vm") - default_vm = marker.args[0] if marker is not None else None - if default_vm is not None: - logging.info(">> No VM specified on CLI. Using default: %s." % default_vm) - ref = default_vm + if marker is not None: + ref = marker.args[0] + logging.info(">> No VM specified on CLI. Using default: %s.", ref) else: # global default logging.info(">> No VM specified on CLI, and no default found in test definition. Using global default.") diff --git a/data.py-dist b/data.py-dist index 8399d8ed1..a7e323611 100644 --- a/data.py-dist +++ b/data.py-dist @@ -1,9 +1,10 @@ # Configuration file, to be adapted to one's needs -from typing import Any, Dict, TYPE_CHECKING +from __future__ import annotations import legacycrypt as crypt # type: ignore import os +from typing import Any, TYPE_CHECKING if TYPE_CHECKING: from lib.typing import IsoImageDef @@ -21,7 +22,7 @@ def hash_password(password): HOST_DEFAULT_PASSWORD_HASH = hash_password(HOST_DEFAULT_PASSWORD) -# Public key for a private key available to the test runner +# Public keys for a private keys available to the test runner TEST_SSH_PUBKEY = """ ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIMnN/wVdQqHA8KsndfrLS7fktH/IEgxoa533efuXR6rw XCP-ng CI ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDKz9uQOoxq6Q0SQ0XTzQHhDolvuo/7EyrDZsYQbRELhcPJG8MT/o5u3HyJFhIP2+HqBSXXgmqRPJUkwz9wUwb2sUwf44qZm/pyPUWOoxyVtrDXzokU/uiaNKUMhbnfaXMz6Ogovtjua63qld2+ZRXnIgrVtYKtYBeu/qKGVSnf4FTOUKl1w3uKkr59IUwwAO8ay3wVnxXIHI/iJgq6JBgQNHbn3C/SpYU++nqL9G7dMyqGD36QPFuqH/cayL8TjNZ67TgAzsPX8OvmRSqjrv3KFbeSlpS/R4enHkSemhgfc8Z2f49tE7qxWZ6x4Uyp5E6ur37FsRf/tEtKIUJGMRXN XCP-ng CI @@ -37,7 +38,7 @@ OBJECTS_NAME_PREFIX = None # skip_xo_config allows to not touch XO's configuration regarding the host # Else the default behaviour is to add the host to XO servers at the beginning # of the testing session and remove it at the end. -HOSTS: Dict[str, Dict[str, Any]] = { +HOSTS: dict[str, dict[str, Any]] = { # "10.0.0.1": {"user": "root", "password": ""}, # "testhost1": {"user": "root", "password": "", 'skip_xo_config': True}, } @@ -107,7 +108,7 @@ OTHER_GUEST_TOOLS = { } # Tools -TOOLS: Dict[str, str] = { +TOOLS: dict[str, str] = { # "iso-remaster": "/home/user/src/xcpng/xcp/scripts/iso-remaster/iso-remaster.sh", } @@ -127,7 +128,7 @@ ISO_IMAGES_CACHE = "/home/user/iso" # for local-only ISO with things like "locally-built/my.iso" or "xs/8.3.iso". # If 'net-only' is set to 'True' only source of type URL will be possible. # By default the parameter is set to False. -ISO_IMAGES: Dict[str, "IsoImageDef"] = { +ISO_IMAGES: dict[str, "IsoImageDef"] = { '83nightly': {'path': os.environ.get("XCPNG83_NIGHTLY", "http://unconfigured.iso"), 'unsigned': True}, @@ -182,25 +183,25 @@ DEFAULT_SR = 'default' CACHE_IMPORTED_VM = False # Default NFS device config: -NFS_DEVICE_CONFIG: Dict[str, Dict[str, str]] = { +NFS_DEVICE_CONFIG: dict[str, dict[str, str]] = { # 'server': '10.0.0.2', # URL/Hostname of NFS server # 'serverpath': '/path/to/shared/mount' # Path to shared mountpoint } # Default NFS4+ only device config: -NFS4_DEVICE_CONFIG: Dict[str, Dict[str, str]] = { +NFS4_DEVICE_CONFIG: dict[str, dict[str, str]] = { # 'server': '10.0.0.2', # URL/Hostname of NFS server # 'serverpath': '/path_to_shared_mount' # Path to shared mountpoint # 'nfsversion': '4.1' } # Default NFS ISO device config: -NFS_ISO_DEVICE_CONFIG: Dict[str, Dict[str, str]] = { +NFS_ISO_DEVICE_CONFIG: dict[str, dict[str, str]] = { # 'location': '10.0.0.2:/path/to/shared/mount' # URL/Hostname of NFS server and path to shared mountpoint } # Default CIFS ISO device config: -CIFS_ISO_DEVICE_CONFIG: Dict[str, Dict[str, str]] = { +CIFS_ISO_DEVICE_CONFIG: dict[str, dict[str, str]] = { # 'location': r'\\10.0.0.2\', # 'username': '', # 'cifspassword': '', @@ -208,18 +209,18 @@ CIFS_ISO_DEVICE_CONFIG: Dict[str, Dict[str, str]] = { # 'vers': '<1.0> or <3.0>' } -CEPHFS_DEVICE_CONFIG: Dict[str, Dict[str, str]] = { +CEPHFS_DEVICE_CONFIG: dict[str, dict[str, str]] = { # 'server': '10.0.0.2', # 'serverpath': '/vms' } -MOOSEFS_DEVICE_CONFIG: Dict[str, Dict[str, str]] = { +MOOSEFS_DEVICE_CONFIG: dict[str, dict[str, str]] = { # 'masterhost': 'mfsmaster', # 'masterport': '9421', # 'rootpath': '/vms' } -LVMOISCSI_DEVICE_CONFIG: Dict[str, Dict[str, str]] = { +LVMOISCSI_DEVICE_CONFIG: dict[str, dict[str, str]] = { # 'target': '192.168.1.1', # 'port': '3260', # 'targetIQN': 'target.example', @@ -248,16 +249,7 @@ BASE_ANSWERFILES = dict( }, ) -IMAGE_EQUIVS: Dict[str, str] = { +IMAGE_EQUIVS: dict[str, str] = { # 'install.test::Nested::install[bios-830-ext]-vm1-607cea0c825a4d578fa5fab56978627d8b2e28bb': # 'install.test::Nested::install[bios-830-ext]-vm1-addb4ead4da49856e1d2fb3ddf4e31027c6b693b', } - -# compatibility settings for older tests -DEFAULT_NFS_DEVICE_CONFIG = NFS_DEVICE_CONFIG -DEFAULT_NFS4_DEVICE_CONFIG = NFS4_DEVICE_CONFIG -DEFAULT_NFS_ISO_DEVICE_CONFIG = NFS_ISO_DEVICE_CONFIG -DEFAULT_CIFS_ISO_DEVICE_CONFIG = CIFS_ISO_DEVICE_CONFIG -DEFAULT_CEPHFS_DEVICE_CONFIG = CEPHFS_DEVICE_CONFIG -DEFAULT_MOOSEFS_DEVICE_CONFIG = MOOSEFS_DEVICE_CONFIG -DEFAULT_LVMOISCSI_DEVICE_CONFIG = LVMOISCSI_DEVICE_CONFIG diff --git a/lib/installer.py b/lib/installer.py index b2e00e3bf..0de688589 100644 --- a/lib/installer.py +++ b/lib/installer.py @@ -19,6 +19,8 @@ def write_xml(self, filename): def top_append(self, *defs): for defn in defs: + if defn is None: + continue self.defn['CONTENTS'].append(self._normalize_structure(defn)) return self @@ -30,15 +32,30 @@ def top_setattr(self, attrs): # makes a mutable deep copy of all `contents` @staticmethod def _normalize_structure(defn): - assert isinstance(defn, dict) - assert 'TAG' in defn - defn = dict(defn) - if 'CONTENTS' not in defn: - defn['CONTENTS'] = [] - if not isinstance(defn['CONTENTS'], str): - defn['CONTENTS'] = [AnswerFile._normalize_structure(item) - for item in defn['CONTENTS']] - return defn + assert isinstance(defn, dict), f"{defn!r} is not a dict" + assert 'TAG' in defn, f"{defn} has no TAG" + + # type mutation through nearly-shallow copy + new_defn = { + 'TAG': defn['TAG'], + 'CONTENTS': [], + } + for key, value in defn.items(): + if key == 'CONTENTS': + if isinstance(value, str): + new_defn['CONTENTS'] = value + else: + new_defn['CONTENTS'] = [ + AnswerFile._normalize_structure(item) + for item in value + if item is not None + ] + elif key == 'TAG': + pass # already copied + else: + new_defn[key] = value + + return new_defn # convert to a ElementTree.Element tree suitable for further # modification before we serialize it to XML @@ -50,14 +67,14 @@ def _defn_to_xml_et(defn, /, *, parent=None): assert isinstance(name, str) contents = defn.pop('CONTENTS', ()) assert isinstance(contents, (str, list)) - element = ET.Element(name, **defn) + element = ET.Element(name, {}, **defn) if parent is not None: parent.append(element) if isinstance(contents, str): element.text = contents else: - for contents in contents: - AnswerFile._defn_to_xml_et(contents, parent=element) + for content in contents: + AnswerFile._defn_to_xml_et(content, parent=element) return element def poweroff(ip): diff --git a/lib/pxe.py b/lib/pxe.py index 4e3491aef..eb94a45c1 100644 --- a/lib/pxe.py +++ b/lib/pxe.py @@ -1,9 +1,11 @@ +from __future__ import annotations + from lib.commands import ssh, scp from data import ARP_SERVER, PXE_CONFIG_SERVER PXE_CONFIG_DIR = "/pxe/configs/custom" -def generate_boot_conf(directory, installer, action): +def generate_boot_conf(directory: str, installer: str, action: str) -> None: # in case of restore, we disable the text ui from the installer completely, # to workaround a bug that leaves us stuck on a confirmation dialog at the end of the operation. rt = 'rt=1' if action == 'restore' else '' @@ -15,7 +17,7 @@ def generate_boot_conf(directory, installer, action): {rt} """) -def server_push_config(mac_address, tmp_local_path): +def server_push_config(mac_address: str, tmp_local_path: str) -> None: assert mac_address remote_dir = f'{PXE_CONFIG_DIR}/{mac_address}/' server_remove_config(mac_address) @@ -23,17 +25,17 @@ def server_push_config(mac_address, tmp_local_path): scp(PXE_CONFIG_SERVER, f'{tmp_local_path}/boot.conf', remote_dir) scp(PXE_CONFIG_SERVER, f'{tmp_local_path}/answerfile.xml', remote_dir) -def server_remove_config(mac_address): +def server_remove_config(mac_address: str) -> None: assert mac_address # protection against deleting the whole parent dir! remote_dir = f'{PXE_CONFIG_DIR}/{mac_address}/' ssh(PXE_CONFIG_SERVER, ['rm', '-rf', remote_dir]) -def server_remove_bootconf(mac_address): +def server_remove_bootconf(mac_address: str) -> None: assert mac_address distant_file = f'{PXE_CONFIG_DIR}/{mac_address}/boot.conf' ssh(PXE_CONFIG_SERVER, ['rm', '-rf', distant_file]) -def arp_addresses_for(mac_address): +def arp_addresses_for(mac_address: str) -> list[str]: output = ssh( ARP_SERVER, ['ip', 'neigh', 'show', 'nud', 'reachable', diff --git a/lib/typing.py b/lib/typing.py index 6b2a7d491..1256de40b 100644 --- a/lib/typing.py +++ b/lib/typing.py @@ -1,5 +1,10 @@ +import sys from typing import TypedDict -from typing_extensions import NotRequired + +if sys.version_info >= (3, 11): + from typing import NotRequired +else: + from typing_extensions import NotRequired IsoImageDef = TypedDict('IsoImageDef', {'path': str, diff --git a/tests/install/conftest.py b/tests/install/conftest.py index 03357c613..90cd1a469 100644 --- a/tests/install/conftest.py +++ b/tests/install/conftest.py @@ -68,13 +68,6 @@ def answerfile(request): answerfile_def = callable_marker(marker.args[0], request) assert isinstance(answerfile_def, AnswerFile) - answerfile_def.top_append( - dict(TAG="admin-interface", - name="eth0", - proto="dhcp", - ), - ) - yield answerfile_def @@ -103,7 +96,7 @@ def installer_iso(request): ) @pytest.fixture(scope='function') -def install_disk(request): +def system_disks_names(request): firmware = request.getfixturevalue("firmware") yield {"uefi": "nvme0n1", "bios": "sda"}[firmware] @@ -179,7 +172,7 @@ def remastered_iso(installer_iso, answerfile): test "$eth_mac" = "$br_mac" fi -if [ $(readlink "/bin/ping") = busybox ]; then +if [ "$(readlink /bin/ping)" = busybox ]; then # XS before 7.0 PINGARGS="" else diff --git a/tests/install/test.py b/tests/install/test.py index daacab631..51ad027fe 100644 --- a/tests/install/test.py +++ b/tests/install/test.py @@ -26,17 +26,18 @@ def helper_vm_with_plugged_disk(running_vm, create_vms): all_vdis = [VDI(uuid, host=host_vm.host) for uuid in host_vm.vdi_uuids()] disk_vdis = [vdi for vdi in all_vdis if not vdi.readonly()] - vdi, = disk_vdis - vbd = helper_vm.create_vbd("1", vdi.uuid) + vbds = [helper_vm.create_vbd(str(n + 1), vdi.uuid) for n, vdi in enumerate(disk_vdis)] try: - vbd.plug() + for vbd in vbds: + vbd.plug() yield helper_vm finally: - vbd.unplug() - vbd.destroy() + for vbd in reversed(vbds): + vbd.unplug() + vbd.destroy() @pytest.mark.dependency() class TestNested: @@ -78,18 +79,19 @@ class TestNested: vifs=[dict(index=0, network_name=NETWORKS["MGMT"])], )) @pytest.mark.answerfile( - lambda install_disk, local_sr, package_source, iso_version: AnswerFile("INSTALL") + lambda system_disks_names, local_sr, package_source, iso_version: AnswerFile("INSTALL") .top_setattr({} if local_sr == "nosr" else {"sr-type": local_sr}) .top_append( - {"TAG": "source", "type": "local"} if package_source == "iso" - else {"TAG": "source", "type": "url", - "CONTENTS": ISO_IMAGES[iso_version]['net-url']} if package_source == "net" - else {}, + {"iso": {"TAG": "source", "type": "local"}, + "net": {"TAG": "source", "type": "url", + "CONTENTS": ISO_IMAGES[iso_version]['net-url']}, + }[package_source], + {"TAG": "admin-interface", "name": "eth0", "proto": "dhcp"}, {"TAG": "primary-disk", "guest-storage": "no" if local_sr == "nosr" else "yes", - "CONTENTS": install_disk}, + "CONTENTS": system_disks_names[0]}, )) - def test_install(self, vm_booted_with_installer, install_disk, + def test_install(self, vm_booted_with_installer, system_disks_names, firmware, iso_version, package_source, local_sr): host_vm = vm_booted_with_installer installer.monitor_install(ip=host_vm.ip) @@ -332,15 +334,15 @@ def test_boot_inst(self, create_vms, vm="vm1", image_test=f"TestNested::test_boot_inst[{firmware}-{orig_version}-{machine}-{package_source}-{local_sr}]")]) @pytest.mark.answerfile( - lambda install_disk, package_source, iso_version: AnswerFile("UPGRADE").top_append( - {"TAG": "source", "type": "local"} if package_source == "iso" - else {"TAG": "source", "type": "url", - "CONTENTS": ISO_IMAGES[iso_version]['net-url']} if package_source == "net" - else {}, + lambda system_disks_names, package_source, iso_version: AnswerFile("UPGRADE").top_append( + {"iso": {"TAG": "source", "type": "local"}, + "net": {"TAG": "source", "type": "url", + "CONTENTS": ISO_IMAGES[iso_version]['net-url']}, + }[package_source], {"TAG": "existing-installation", - "CONTENTS": install_disk}, + "CONTENTS": system_disks_names[0]}, )) - def test_upgrade(self, vm_booted_with_installer, install_disk, + def test_upgrade(self, vm_booted_with_installer, system_disks_names, firmware, orig_version, iso_version, machine, package_source, local_sr): host_vm = vm_booted_with_installer installer.monitor_upgrade(ip=host_vm.ip) @@ -393,11 +395,11 @@ def test_boot_upg(self, create_vms, vm="vm1", image_test=f"TestNested::test_boot_upg[{firmware}-{orig_version}-host1-{package_source}-{local_sr}]")]) @pytest.mark.answerfile( - lambda install_disk: AnswerFile("RESTORE").top_append( + lambda system_disks_names: AnswerFile("RESTORE").top_append( {"TAG": "backup-disk", - "CONTENTS": install_disk}, + "CONTENTS": system_disks_names[0]}, )) - def test_restore(self, vm_booted_with_installer, install_disk, + def test_restore(self, vm_booted_with_installer, system_disks_names, firmware, orig_version, iso_version, package_source, local_sr): host_vm = vm_booted_with_installer installer.monitor_restore(ip=host_vm.ip)