From 6cfb5a7b6ba2d15bfd898f2bdaf8faf5006235cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=ABtan=20Lehmann?= Date: Wed, 28 May 2025 16:08:22 +0200 Subject: [PATCH 1/5] fix typehints in install tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit note that `with_args` is used as a workaround for a pytest type-hint bug that declares the function passed to a marker as returned by the marker, which is not true with a lambda. Signed-off-by: Gaëtan Lehmann --- tests/install/conftest.py | 2 ++ tests/install/test.py | 17 +++++++++-------- tests/install/test_fixtures.py | 2 +- 3 files changed, 12 insertions(+), 9 deletions(-) diff --git a/tests/install/conftest.py b/tests/install/conftest.py index 90cd1a469..2262adf2a 100644 --- a/tests/install/conftest.py +++ b/tests/install/conftest.py @@ -1,5 +1,6 @@ import logging import os +from typing import Sequence import pytest import pytest_dependency # type: ignore import tempfile @@ -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'], diff --git a/tests/install/test.py b/tests/install/test.py index 51ad027fe..b17ef8c6b 100644 --- a/tests/install/test.py +++ b/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}]"))]) diff --git a/tests/install/test_fixtures.py b/tests/install/test_fixtures.py index 0d6247693..53d08a50b 100644 --- a/tests/install/test_fixtures.py +++ b/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"}, )) From 59ffd81e0298577f05567da4614bcbaa5cb9bbc3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=ABtan=20Lehmann?= Date: Wed, 28 May 2025 17:21:25 +0200 Subject: [PATCH 2/5] ensure_type: a method to cast a type and check the type at runtime MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Gaëtan Lehmann --- lib/common.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/lib/common.py b/lib/common.py index 2c8b36e91..69aa35fac 100644 --- a/lib/common.py +++ b/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 Any, Dict, Literal, Optional, Type, TypeVar, overload, TYPE_CHECKING, Union from uuid import UUID import pytest @@ -67,6 +67,14 @@ 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,))) +T = TypeVar("T") + +def ensure_type(typ: Type[T], value: Any) -> T: + """Converts a value to the specified type. Also performs a runtime check.""" + if not isinstance(value, typ): + raise TypeError(f"'{type(value).__name__}' object is not of the expected type '{typ.__name__}'") + return value + def callable_marker(value, request): """ Process value optionally generated by fixture-dependent callable. From 203fc7711ba05640c7015d486399867390babac8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=ABtan=20Lehmann?= Date: Thu, 29 May 2025 10:50:27 +0200 Subject: [PATCH 3/5] ensure_type: supports complex types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit like TypeDict Signed-off-by: Gaëtan Lehmann --- lib/common.py | 21 +++++++++++++++++---- requirements/base.txt | 1 + 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/lib/common.py b/lib/common.py index 69aa35fac..f406c7d30 100644 --- a/lib/common.py +++ b/lib/common.py @@ -1,3 +1,4 @@ +from contextlib import suppress import getpass import inspect import itertools @@ -7,6 +8,7 @@ import time 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 uuid import UUID @@ -70,10 +72,21 @@ def expand_scope_relative_nodeid(scoped_nodeid, scope, ref_nodeid): T = TypeVar("T") def ensure_type(typ: Type[T], value: Any) -> T: - """Converts a value to the specified type. Also performs a runtime check.""" - if not isinstance(value, typ): - raise TypeError(f"'{type(value).__name__}' object is not of the expected type '{typ.__name__}'") - return value + """ + 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 = 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, request): """ diff --git a/requirements/base.txt b/requirements/base.txt index ecfd49e51..82eeb01df 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -6,3 +6,4 @@ pluggy>=1.1.0 requests legacycrypt pytest-dependency +pydantic From 1d93cc81563e793d662c12b0b15c8775aa48f2cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=ABtan=20Lehmann?= Date: Thu, 29 May 2025 10:50:27 +0200 Subject: [PATCH 4/5] ensure_type: cache the type adaptors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit for a much more efficient type checking of complex types (using pydantic) at runtime: before: 124 μs ± 82.5 ns per loop (mean ± std. dev. of 7 runs, 10,000 loops each) after: 1.81 μs ± 4.74 ns per loop (mean ± std. dev. of 7 runs, 1,000,000 loops each) for comparison, when using isinstance: 629 ns ± 9.26 ns per loop (mean ± std. dev. of 7 runs, 1,000,000 loops each) Signed-off-by: Gaëtan Lehmann --- lib/common.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/lib/common.py b/lib/common.py index f406c7d30..f57e78bc9 100644 --- a/lib/common.py +++ b/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, Dict, Literal, Optional, Type, TypeVar, cast, overload, TYPE_CHECKING, Union from uuid import UUID import pytest @@ -71,6 +71,8 @@ def expand_scope_relative_nodeid(scoped_nodeid, scope, ref_nodeid): 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. @@ -84,7 +86,7 @@ def ensure_type(typ: Type[T], value: Any) -> T: except TypeError: # not just a simple type, lets try with pydantic with suppress(ValidationError): - ta = TypeAdapter(typ) + 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__}'") From 36c3085a9597e1341348540029b8bf7fc429ffb9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=ABtan=20Lehmann?= Date: Wed, 28 May 2025 17:21:25 +0200 Subject: [PATCH 5/5] callable_marker: add type hints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Gaëtan Lehmann Co-Authored-By: Yann Dirson --- lib/common.py | 12 ++++++++---- tests/install/conftest.py | 16 +++++++++------- 2 files changed, 17 insertions(+), 11 deletions(-) diff --git a/lib/common.py b/lib/common.py index f57e78bc9..6b8090031 100644 --- a/lib/common.py +++ b/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, cast, 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 @@ -90,7 +90,7 @@ def ensure_type(typ: Type[T], value: Any) -> T: return ta.validate_python(value) raise TypeError(f"'{type(value).__name__}' object is not of the expected type '{typ.__name__}'") -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. @@ -106,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: diff --git a/tests/install/conftest.py b/tests/install/conftest.py index 2262adf2a..a8c5b42ac 100644 --- a/tests/install/conftest.py +++ b/tests/install/conftest.py @@ -1,13 +1,15 @@ +from __future__ import annotations + import logging import os -from typing import Sequence +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 @@ -39,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. @@ -66,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 @@ -322,8 +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) - assert isinstance(continuation_of, Sequence) + 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'],