installer: typecheck of AnswerFile dict

Note the `type: ignore[call-arg]` in test_fixtures is here to overcome a
bug/limitation in mypy 1.15.0: in callable_marker() it seems to consider
that `value(**params)` produces at least one argument.

Signed-off-by: Yann Dirson <[email protected]>

Author: ydirson

data.py-dist

@@ -7,7 +7,7 @@ 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.
@@ -227,7 +227,7 @@ LVMOISCSI_DEVICE_CONFIG: dict[str, dict[str, str]] = {
 #    'SCSIid': 'id'
 }
 
-BASE_ANSWERFILES = dict(
+BASE_ANSWERFILES: dict[str, "SimpleAnswerfileDict"] = dict(
     INSTALL={
         "TAG": "installation",
         "CONTENTS": (

lib/installer.py

@@ -1,42 +1,47 @@
+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):
+    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 = {
+        new_defn: AnswerfileDict = {
             'TAG': defn['TAG'],
             'CONTENTS': [],
         }
@@ -45,29 +50,37 @@ def _normalize_structure(defn):
                 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
+                        for item in value_as_sequence
                         if item is not None
                     ]
             elif key == 'TAG':
                 pass            # already copied
             else:
-                new_defn[key] = value
+                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):
@@ -77,7 +90,7 @@ def _defn_to_xml_et(defn, /, *, parent=None):
                 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:
@@ -88,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'",
@@ -112,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'",
@@ -145,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/typing.py

@@ -1,14 +1,39 @@
+from __future__ import annotations
+
 import sys
-from typing import TypedDict
+from typing import Sequence, TypedDict, Union
 
 if sys.version_info >= (3, 11):
-    from typing import NotRequired
+    from typing import NotRequired, Self
 else:
-    from typing_extensions import NotRequired
+    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.
+    'guest-storage': NotRequired[str],
+    'mode': NotRequired[str],
+    'name': NotRequired[str],
+    'proto': NotRequired[str],
+    'type': NotRequired[str],
+})

tests/install/test_fixtures.py

@@ -5,7 +5,7 @@
 
 # test the answerfile fixture can run on 2 parametrized instances
 # of the test in one run
-@pytest.mark.answerfile(lambda: AnswerFile("INSTALL").top_append(
+@pytest.mark.answerfile(lambda: AnswerFile("INSTALL").top_append( # type: ignore[call-arg]
     {"TAG": "source", "type": "local"},
     {"TAG": "primary-disk", "CONTENTS": "nvme0n1"},
 ))