Installer features prerequisites

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.")

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,44 +183,44 @@ 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\<shared folder name>',
 #    'username': '<user>',
 #    'cifspassword': '<password>',
 #    'type': 'cifs',
 #    '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

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):

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,25 +17,25 @@ 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)
     ssh(PXE_CONFIG_SERVER, ['mkdir', '-p', remote_dir])
     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',

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,

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

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)