Test installer's IPv6 features

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,12 +1,13 @@
 # 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
+    from lib.typing import SimpleAnswerfileDict, IsoImageDef
 
 # Default user and password to connect to a host through XAPI
 # Note: this won't be used for SSH.
@@ -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},
 }
@@ -106,8 +107,25 @@ OTHER_GUEST_TOOLS = {
     },
 }
 
+# IP addresses for hosts to be installed
+# NOTE: do NOT set an IP for host1, it is assumed to use DEFAULT
+HOSTS_IP_CONFIG: dict[str, dict[str, str]] = {
+    'HOSTS': {
+#        'DEFAULT': '192.16.0.1',
+#        'host2': '192.16.0.2',
+#        "DEFAULT_v6": '2001:db8::1:1/32',
+#        "host_v6": '2001:db8::1:2/32',
+    },
+#    'NETMASK': '255.255.0.0',
+#    'GATEWAY': '192.16.0.254',
+#    'DNS': '192.16.0.254',
+
+#    'GATEWAY_v6': '2001:db8::1',
+#    'DNS_v6': '2001:db8::2',
+}
+
 # Tools
-TOOLS: Dict[str, str] = {
+TOOLS: dict[str, str] = {
 #    "iso-remaster": "/home/user/src/xcpng/xcp/scripts/iso-remaster/iso-remaster.sh",
 }
 
@@ -127,7 +145,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,51 +200,51 @@ 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',
 #    'SCSIid': 'id'
 }
 
-BASE_ANSWERFILES = dict(
+BASE_ANSWERFILES: dict[str, "SimpleAnswerfileDict"] = dict(
     INSTALL={
         "TAG": "installation",
         "CONTENTS": (
@@ -248,16 +266,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

@@ -1,66 +1,96 @@
+from __future__ import annotations
+
 import logging
 import time
 import xml.etree.ElementTree as ET
+from typing import cast, Optional, Sequence, Union
 
 from lib.commands import ssh, SSHCommandFailed
 from lib.common import wait_for
+from lib.typing import AnswerfileDict, Self, SimpleAnswerfileDict
 
 class AnswerFile:
-    def __init__(self, kind, /):
+    def __init__(self, kind: str, /) -> None:
         from data import BASE_ANSWERFILES
-        defn = BASE_ANSWERFILES[kind]
+        defn: SimpleAnswerfileDict = BASE_ANSWERFILES[kind]
         self.defn = self._normalize_structure(defn)
 
-    def write_xml(self, filename):
+    def write_xml(self, filename: str) -> None:
         etree = ET.ElementTree(self._defn_to_xml_et(self.defn))
         etree.write(filename)
 
     # chainable mutators for lambdas
 
-    def top_append(self, *defs):
+    def top_append(self, *defs: Union[SimpleAnswerfileDict, None, ValueError]) -> Self:
+        assert not isinstance(self.defn['CONTENTS'], str), "a toplevel CONTENTS must be a list"
         for defn in defs:
+            if defn is None:
+                continue
             self.defn['CONTENTS'].append(self._normalize_structure(defn))
         return self
 
-    def top_setattr(self, attrs):
+    def top_setattr(self, attrs: "dict[str, str]") -> Self:
         assert 'CONTENTS' not in attrs
-        self.defn.update(attrs)
+        self.defn.update(cast(AnswerfileDict, attrs))
         return self
 
     # 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
+    def _normalize_structure(defn: Union[SimpleAnswerfileDict, ValueError]) -> AnswerfileDict:
+        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: AnswerfileDict = {
+            'TAG': defn['TAG'],
+            'CONTENTS': [],
+        }
+        for key, value in defn.items():
+            if key == 'CONTENTS':
+                if isinstance(value, str):
+                    new_defn['CONTENTS'] = value
+                else:
+                    value_as_sequence: Sequence["SimpleAnswerfileDict"]
+                    if isinstance(value, Sequence):
+                        value_as_sequence = value
+                    else:
+                        value_as_sequence = (
+                            cast(SimpleAnswerfileDict, value),
+                        )
+                    new_defn['CONTENTS'] = [
+                        AnswerFile._normalize_structure(item)
+                        for item in value_as_sequence
+                        if item is not None
+                    ]
+            elif key == 'TAG':
+                pass            # already copied
+            else:
+                new_defn[key] = value # type: ignore[literal-required]
+
+        return new_defn
 
     # convert to a ElementTree.Element tree suitable for further
     # modification before we serialize it to XML
     @staticmethod
-    def _defn_to_xml_et(defn, /, *, parent=None):
+    def _defn_to_xml_et(defn: AnswerfileDict, /, *, parent: Optional[ET.Element] = None) -> ET.Element:
         assert isinstance(defn, dict)
-        defn = dict(defn)
-        name = defn.pop('TAG')
+        defn_copy = dict(defn)
+        name = defn_copy.pop('TAG')
         assert isinstance(name, str)
-        contents = defn.pop('CONTENTS', ())
+        contents = cast(Union[str, "list[AnswerfileDict]"], defn_copy.pop('CONTENTS', []))
         assert isinstance(contents, (str, list))
-        element = ET.Element(name, **defn)
+        defn_filtered = cast("dict[str, str]", defn_copy)
+        element = ET.Element(name, {}, **defn_filtered)
         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):
+def poweroff(ip: str) -> None:
     try:
         ssh(ip, ["poweroff"])
     except SSHCommandFailed as e:
@@ -71,7 +101,7 @@ def poweroff(ip):
         else:
             raise
 
-def monitor_install(*, ip):
+def monitor_install(*, ip: str) -> None:
     # wait for "yum install" phase to finish
     wait_for(lambda: ssh(ip, ["grep",
                               "'DISPATCH: NEW PHASE: Completing installation'",
@@ -95,7 +125,7 @@ def monitor_install(*, ip):
                          ).returncode == 1,
              "Wait for installer to terminate")
 
-def monitor_upgrade(*, ip):
+def monitor_upgrade(*, ip: str) -> None:
     # wait for "yum install" phase to start
     wait_for(lambda: ssh(ip, ["grep",
                               "'DISPATCH: NEW PHASE: Reading package information'",
@@ -128,7 +158,7 @@ def monitor_upgrade(*, ip):
                          ).returncode == 1,
              "Wait for installer to terminate")
 
-def monitor_restore(*, ip):
+def monitor_restore(*, ip: str) -> None:
     # wait for "yum install" phase to start
     wait_for(lambda: ssh(ip, ["grep",
                               "'Restoring backup'",

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,9 +1,41 @@
-from typing import TypedDict
-from typing_extensions import NotRequired
+from __future__ import annotations
+
+import sys
+from typing import Sequence, TypedDict, Union
+
+if sys.version_info >= (3, 11):
+    from typing import NotRequired, Self
+else:
+    from typing_extensions import NotRequired, Self
 
 IsoImageDef = TypedDict('IsoImageDef',
                         {'path': str,
                          'net-url': NotRequired[str],
                          'net-only': NotRequired[bool],
                          'unsigned': NotRequired[bool],
                          })
+
+
+# Dict-based description of an Answerfile object to be built.
+AnswerfileDict = TypedDict('AnswerfileDict', {
+    'TAG': str,
+    'CONTENTS': Union[str, "list[AnswerfileDict]"],
+})
+
+# Simplified version of AnswerfileDict for user input.
+# - does not require to write 0 or 1 subelement as a list
+SimpleAnswerfileDict = TypedDict('SimpleAnswerfileDict', {
+    'TAG': str,
+    'CONTENTS': NotRequired[Union[str, "SimpleAnswerfileDict", Sequence["SimpleAnswerfileDict"]]],
+
+    # No way to allow arbitrary fields in addition?  This conveys the
+    # field's type, but allows them in places we wouldn't want them,
+    # and forces every XML attribute we use to appear here.
+    'device': NotRequired[str],
+    'guest-storage': NotRequired[str],
+    'mode': NotRequired[str],
+    'name': NotRequired[str],
+    'proto': NotRequired[str],
+    'protov6': NotRequired[str],
+    'type': NotRequired[str],
+})