callable_marker: add type hints

Signed-off-by: Gaëtan Lehmann <[email protected]>

Author: glehmann

lib/common.py

@@ -9,7 +9,7 @@
 import traceback
 from enum import Enum
 from pydantic import TypeAdapter, ValidationError
-from typing import Any, Dict, Literal, Optional, Type, TypeVar, overload, TYPE_CHECKING, Union
+from typing import Any, Callable, Dict, Literal, Optional, Type, TypeVar, cast, overload, TYPE_CHECKING, Union
 from uuid import UUID
 
 import pytest
@@ -69,7 +69,6 @@ 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] = {}
@@ -97,6 +96,7 @@ def ensure_type(typ: Type[T], value: Any) -> T:
         raise TypeError(f"'{type(value).__name__}' object is not of the expected type '{typ.__name__}'")
     return value
 
+def callable_marker(value: Union[T, Callable[[], T]], request: pytest.FixtureRequest) -> T:
     """
     Process value optionally generated by fixture-dependent callable.
 
@@ -112,8 +112,11 @@ def ensure_type(typ: Type[T], value: Any) -> T:
                       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
+        return cast(Callable[[], T], value)(**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,13 +1,13 @@
 import logging
 import os
-from typing import Sequence
+from typing import Callable, 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
 
@@ -60,14 +60,14 @@ def answerfile(request):
     >     answerfile.write_xml("my-answers.xml")
     """
     marker = request.node.get_closest_marker("answerfile")
+    marker_args = ensure_type(Sequence[Union[AnswerFile, Callable[[], AnswerFile]]], marker.args)
 
     if marker is None:
         yield None              # no answerfile to generate
         return
 
     # construct answerfile definition from option "base", and explicit bits
-    answerfile_def = callable_marker(marker.args[0], request)
-    assert isinstance(answerfile_def, AnswerFile)
+    answerfile_def = callable_marker(marker_args[0], request)
 
     answerfile_def.top_append(
         dict(TAG="admin-interface",
@@ -329,8 +329,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)
-    assert isinstance(continuation_of, Sequence)
+    marker_args = ensure_type(Sequence[Union[Sequence[dict], Callable[[], Sequence[dict]]]], marker.args)
+    continuation_of = callable_marker(marker_args[0], request)
 
     vm_defs = [dict(name=vm_spec['vm'],
                     image_test=vm_spec['image_test'],