Merge pull request #323 from xcp-ng/typing-fixes

Typing fixes

Author: ydirson

.github/workflows/code-checkers.yml

@@ -19,7 +19,7 @@ jobs:
     - name: Create a dummy data.py
       run: cp data.py-dist data.py
     - name: Install additional typing data and check with mypy
-      run: mypy --install-types --non-interactive lib/ tests/
+      run: mypy --install-types --non-interactive lib/ conftest.py pkgfixtures.py tests/
 
   pyright:
     runs-on: ubuntu-latest
@@ -37,4 +37,4 @@ jobs:
     - name: Create a dummy data.py
       run: cp data.py-dist data.py
     - name: Check with pyright
-      run: pyright lib/ # tests/
+      run: pyright lib/ conftest.py pkgfixtures.py # tests/

conftest.py

@@ -544,10 +544,10 @@ def create_vms(request, host, tests_git_revision):
         # FIXME should check for extra args
         vm_defs.append(vm_def)
 
+    vms = []
+    vdis = []
+    vbds = []
     try:
-        vms = []
-        vdis = []
-        vbds = []
         for vm_def in vm_defs:
             if "template" in vm_def:
                 _create_vm(request, vm_def, host, vms, vdis, vbds)

lib/common.py

@@ -7,7 +7,7 @@
 import time
 import traceback
 from enum import Enum
-from typing import Dict, Literal, Optional, overload, TYPE_CHECKING, Union
+from typing import Callable, cast, Dict, Literal, Optional, overload, TYPE_CHECKING, TypeVar, Union
 from uuid import UUID
 
 import pytest
@@ -17,6 +17,8 @@
 if TYPE_CHECKING:
     import lib.host
 
+T = TypeVar("T")
+
 class PackageManagerEnum(Enum):
     UNKNOWN = 1
     RPM = 2
@@ -67,7 +69,7 @@ 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):
+def callable_marker(value: Union[T, Callable[..., T]], request: pytest.FixtureRequest) -> T:
     """
     Process value optionally generated by fixture-dependent callable.
 
@@ -83,8 +85,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:

tests/install/conftest.py

@@ -1,5 +1,6 @@
 import logging
 import os
+from typing import Generator, Sequence, Union
 import pytest
 import pytest_dependency        # type: ignore
 import tempfile
@@ -38,7 +39,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,7 +66,7 @@ def answerfile(request):
         return
 
     # construct answerfile definition from option "base", and explicit bits
-    answerfile_def = callable_marker(marker.args[0], request)
+    answerfile_def: AnswerFile = callable_marker(marker.args[0], request)
     assert isinstance(answerfile_def, AnswerFile)
 
     yield answerfile_def
@@ -322,6 +323,7 @@ def xcpng_chained(request):
     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)
+    assert isinstance(continuation_of, Sequence)
 
     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"},
 ))