fix typehints in install tests

lib/common.py

@@ -1,3 +1,4 @@
+from contextlib import suppress
 import getpass
 import inspect
 import itertools
@@ -7,7 +8,8 @@
 import time
 import traceback
 from enum import Enum
-from typing import Dict, Literal, Optional, overload, TYPE_CHECKING, Union
+from pydantic import TypeAdapter, ValidationError
+from typing import Any, Callable, Dict, Literal, Optional, Type, TypeVar, cast, overload, TYPE_CHECKING, Union
 from uuid import UUID
 
 import pytest
@@ -67,7 +69,28 @@ def expand_scope_relative_nodeid(scoped_nodeid, scope, ref_nodeid):
     logging.debug("scope: %r base: %r relative: %r", scope, base, scoped_nodeid)
     return "::".join(itertools.chain(base, (scoped_nodeid,)))
 
-def callable_marker(value, request):
+T = TypeVar("T")
+
+_ensure_type_cache: Dict[type, TypeAdapter] = {}
+
+def ensure_type(typ: Type[T], value: Any) -> T:
+    """
+    Converts a value to the specified type.
+
+    Unlike typing.cast, it also performs a runtime check.
+    Unlike isinstance, it also supports complex types.
+    """
+    try:
+        if isinstance(value, typ):
+            return value
+    except TypeError:
+        # not just a simple type, lets try with pydantic
+        with suppress(ValidationError):
+            ta = cast(TypeAdapter[T], _ensure_type_cache.setdefault(typ, TypeAdapter(typ)))
+            return ta.validate_python(value)
+    raise TypeError(f"'{type(value).__name__}' object is not of the expected type '{typ.__name__}'")
+
+def callable_marker(value: Union[T, Callable[..., T]], request: pytest.FixtureRequest) -> T:
     """
     Process value optionally generated by fixture-dependent callable.
 
@@ -83,8 +106,12 @@ def callable_marker(value, request):
                       for arg_name in inspect.getfullargspec(value).args}
         except pytest.FixtureLookupError as e:
             raise RuntimeError("fixture in mapping not found on test") from e
-        value = value(**params)
-    return value
+        # callable ensures the value is of type Callable[..., object], which is not enough in that case
+        # we can trust the static checker though, and thus use cast
+        fn = cast(Callable[..., T], value)
+        return fn(**params)
+    else:
+        return value
 
 def wait_for(fn, msg=None, timeout_secs=2 * 60, retry_delay_secs=2, invert=False):
     if msg is not None:

requirements/base.txt

@@ -6,3 +6,4 @@ pluggy>=1.1.0
 requests
 legacycrypt
 pytest-dependency
+pydantic

tests/install/conftest.py

@@ -1,12 +1,15 @@
+from __future__ import annotations
+
 import logging
 import os
+from typing import Callable, Generator, Sequence, Union
 import pytest
 import pytest_dependency        # type: ignore
 import tempfile
 import xml.etree.ElementTree as ET
 
 from lib import installer, pxe
-from lib.common import callable_marker, url_download, wait_for
+from lib.common import callable_marker, ensure_type, url_download, wait_for
 from lib.installer import AnswerFile
 from lib.commands import local_cmd
 
@@ -38,7 +41,7 @@ def skip_package_source(version, package_source):
     return True, "unknown source type {}".format(package_source)
 
 @pytest.fixture(scope='function')
-def answerfile(request):
+def answerfile(request: pytest.FixtureRequest) -> Generator[Union[AnswerFile, None], None, None]:
     """
     Makes an AnswerFile object available to test and other fixtures.
 
@@ -65,8 +68,8 @@ def answerfile(request):
         return
 
     # construct answerfile definition from option "base", and explicit bits
-    answerfile_def = callable_marker(marker.args[0], request)
-    assert isinstance(answerfile_def, AnswerFile)
+    value = callable_marker(marker.args[0], request)
+    answerfile_def = ensure_type(AnswerFile, value)
 
     yield answerfile_def
 
@@ -321,7 +324,8 @@ def xcpng_chained(request):
     # take test name from mark
     marker = request.node.get_closest_marker("continuation_of")
     assert marker is not None, "xcpng_chained fixture requires 'continuation_of' marker"
-    continuation_of = callable_marker(marker.args[0], request)
+    value = callable_marker(marker.args[0], request)
+    continuation_of = ensure_type(Sequence[dict], value)
 
     vm_defs = [dict(name=vm_spec['vm'],
                     image_test=vm_spec['image_test'],

tests/install/test.py

@@ -78,7 +78,7 @@ class TestNested:
             cd_vbd=dict(device="xvdd", userdevice="3"),
             vifs=[dict(index=0, network_name=NETWORKS["MGMT"])],
         ))
-    @pytest.mark.answerfile(
+    @pytest.mark.answerfile.with_args(
         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(
@@ -111,7 +111,7 @@ def test_install(self, vm_booted_with_installer, system_disks_names,
         "xs70",
     ))
     @pytest.mark.parametrize("firmware", ("uefi", "bios"))
-    @pytest.mark.continuation_of(
+    @pytest.mark.continuation_of.with_args(
         lambda version, firmware, local_sr, package_source: [dict(
             vm="vm1",
             image_test=f"TestNested::test_install[{firmware}-{version}-{package_source}-{local_sr}]")])
@@ -245,6 +245,7 @@ def _test_firstboot(self, create_vms, mode, *, machine='DEFAULT', is_restore=Fal
                 raise AssertionError(f"Unhandled LSB release {lsb_rel!r}")
             # check for firstboot issues
             # FIXME: flaky, must check logs extraction on failure
+            stamp = ''
             try:
                 for stamp in sorted(STAMPS):
                     wait_for(lambda: pool.master.ssh(["test", "-e", f"{STAMPS_DIR}/{stamp}"],
@@ -302,7 +303,7 @@ def _test_firstboot(self, create_vms, mode, *, machine='DEFAULT', is_restore=Fal
         "xs70",
     ))
     @pytest.mark.parametrize("firmware", ("uefi", "bios"))
-    @pytest.mark.continuation_of(
+    @pytest.mark.continuation_of.with_args(
         lambda firmware, version, machine, local_sr, package_source: [
             dict(vm="vm1",
                  image_test=("TestNested::test_tune_firstboot"
@@ -329,11 +330,11 @@ def test_boot_inst(self, create_vms,
         ("821.1", "821.1"),
     ])
     @pytest.mark.parametrize("firmware", ("uefi", "bios"))
-    @pytest.mark.continuation_of(
+    @pytest.mark.continuation_of.with_args(
         lambda firmware, orig_version, machine, package_source, local_sr: [dict(
             vm="vm1",
             image_test=f"TestNested::test_boot_inst[{firmware}-{orig_version}-{machine}-{package_source}-{local_sr}]")])
-    @pytest.mark.answerfile(
+    @pytest.mark.answerfile.with_args(
         lambda system_disks_names, package_source, iso_version: AnswerFile("UPGRADE").top_append(
             {"iso": {"TAG": "source", "type": "local"},
              "net": {"TAG": "source", "type": "url",
@@ -365,7 +366,7 @@ def test_upgrade(self, vm_booted_with_installer, system_disks_names,
         "821.1-821.1",
     ))
     @pytest.mark.parametrize("firmware", ("uefi", "bios"))
-    @pytest.mark.continuation_of(
+    @pytest.mark.continuation_of.with_args(
         lambda firmware, mode, machine, package_source, local_sr: [dict(
             vm="vm1",
             image_test=(f"TestNested::test_upgrade[{firmware}-{mode}-{machine}-{package_source}-{local_sr}]"))])
@@ -390,7 +391,7 @@ def test_boot_upg(self, create_vms,
         ("821.1-821.1", "821.1"),
     ])
     @pytest.mark.parametrize("firmware", ("uefi", "bios"))
-    @pytest.mark.continuation_of(
+    @pytest.mark.continuation_of.with_args(
         lambda firmware, orig_version, local_sr, package_source: [dict(
             vm="vm1",
             image_test=f"TestNested::test_boot_upg[{firmware}-{orig_version}-host1-{package_source}-{local_sr}]")])
@@ -421,7 +422,7 @@ def test_restore(self, vm_booted_with_installer, system_disks_names,
         "821.1-821.1-821.1",
     ))
     @pytest.mark.parametrize("firmware", ("uefi", "bios"))
-    @pytest.mark.continuation_of(
+    @pytest.mark.continuation_of.with_args(
         lambda firmware, mode, package_source, local_sr: [dict(
             vm="vm1",
             image_test=(f"TestNested::test_restore[{firmware}-{mode}-{package_source}-{local_sr}]"))])

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.with_args(lambda: AnswerFile("INSTALL").top_append(
     {"TAG": "source", "type": "local"},
     {"TAG": "primary-disk", "CONTENTS": "nvme0n1"},
 ))