From: Aleš Mrázek Date: Mon, 16 Feb 2026 22:06:06 +0000 (+0100) Subject: python: modeling: types: added base types X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=f878221e91ff3bb80c65235571fdf2660dc1e86b;p=thirdparty%2Fknot-resolver.git python: modeling: types: added base types --- diff --git a/python/knot_resolver/utils/modeling/types/__init__.py b/python/knot_resolver/utils/modeling/types/__init__.py new file mode 100644 index 000000000..61a7cc3f9 --- /dev/null +++ b/python/knot_resolver/utils/modeling/types/__init__.py @@ -0,0 +1,6 @@ +from .base_generic_custom_types import ListOrItem, Transformed + +__all__ = [ + "ListOrItem", + "Transformed", +] diff --git a/python/knot_resolver/utils/modeling/types/base_float_types.py b/python/knot_resolver/utils/modeling/types/base_float_types.py new file mode 100644 index 000000000..de1c04155 --- /dev/null +++ b/python/knot_resolver/utils/modeling/types/base_float_types.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from knot_resolver.utils.modeling.context import Strictness +from knot_resolver.utils.modeling.errors import DataTypeError, DataValueError + +from .base_custom_type import BaseCustomType + +if TYPE_CHECKING: + from knot_resolver.utils.modeling.context import Context + + +class BaseFloat(BaseCustomType): + """Base class to work with float value.""" + + def _validate(self, context: Context) -> None: + if ( + context.strictness > Strictness.PERMISSIVE + and not isinstance(self._value, (float, int)) + or isinstance(self._value, bool) + ): + msg = ( + f"Unexpected value for '{type(self)}'." + f" Expected float, got '{self._value}' with type '{type(self._value)}'" + ) + raise DataTypeError(msg, self._tree_path) + + def __int__(self) -> int: + return int(self._value) + + def __float__(self) -> float: + return float(self._value) + + @classmethod + def json_schema(cls) -> dict[Any, Any]: + return {"type": "number"} + + +class BaseFloatRange(BaseFloat): + _min: float + _max: float + + def _validate(self, context: Context) -> None: + super()._validate(context) + if context.strictness > Strictness.PERMISSIVE and hasattr(self, "_min") and (self._value < self._min): + msg = f"value {self._value} is lower than the minimum {self._min}." + raise DataValueError(msg, self._tree_path) + if context.strictness > Strictness.PERMISSIVE and hasattr(self, "_max") and (self._value > self._max): + msg = f"value {self._value} is higher than the maximum {self._max}" + raise DataValueError(msg, self._tree_path) + + @classmethod + def json_schema(cls) -> dict[Any, Any]: + typ: dict[str, Any] = {"type": "number"} + if hasattr(cls, "_min"): + typ["minimum"] = cls._min + if hasattr(cls, "_max"): + typ["maximum"] = cls._max + return typ diff --git a/python/knot_resolver/utils/modeling/types/base_generic_custom_types.py b/python/knot_resolver/utils/modeling/types/base_generic_custom_types.py new file mode 100644 index 000000000..60649650e --- /dev/null +++ b/python/knot_resolver/utils/modeling/types/base_generic_custom_types.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from typing import Any, Generic, Iterator, List, TypeVar, Union + +try: + from typing import Annotated +except ImportError: + from typing_extensions import Annotated + +# The type is used to annotate the result and input types for a value (Transformed[resultT, inputT]). +# In order to transform the input into the result value, a transformation method of DataModelNone class is required. +Transformed: type = Annotated + +T = TypeVar("T") + + +class BaseGenericCustomTypeWrapper(Generic[T]): + def __init__(self, value: Any) -> None: + self._value = value + + def __repr__(self) -> str: + return f'{type(self).__name__}("{self._value!r}")' + + def __str__(self) -> str: + return str(self._value) + + def __eq__(self, o: object) -> bool: + if not isinstance(o, type(self)): + return NotImplemented + return self._value == o._value + + +class ListOrItem(BaseGenericCustomTypeWrapper[Union[List[T], T]]): + def _get_list(self) -> list[T]: + return self._value if isinstance(self._value, list) else [self._value] + + def __iter__(self) -> Iterator[T]: + return iter(self._get_list()) diff --git a/python/knot_resolver/utils/modeling/types/base_integer_types.py b/python/knot_resolver/utils/modeling/types/base_integer_types.py new file mode 100644 index 000000000..2375d5b6e --- /dev/null +++ b/python/knot_resolver/utils/modeling/types/base_integer_types.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from knot_resolver.utils.modeling.context import Strictness +from knot_resolver.utils.modeling.errors import DataTypeError, DataValueError + +from .base_custom_type import BaseCustomType + +if TYPE_CHECKING: + from knot_resolver.utils.modeling.context import Context + + +class BaseInteger(BaseCustomType): + """Base class to work with integer value.""" + + def _validate(self, context: Context) -> None: + if ( + context.strictness > Strictness.PERMISSIVE + and not isinstance(self._value, int) + or isinstance(self._value, bool) + ): + msg = ( + f"Unexpected value for '{type(self)}'" + f" Expected integer, got '{self._value}' with type '{type(self._value)}'" + ) + raise DataTypeError(msg, self._tree_path) + + def __int__(self) -> int: + return int(self._value) + + @classmethod + def from_string(cls, value: str, tree_path: str = "/") -> BaseInteger: + try: + return cls(int(value), tree_path) + except ValueError as e: + msg = f"invalid integer {value}" + raise DataValueError(msg) from e + + @classmethod + def json_schema(cls) -> dict[Any, Any]: + return {"type": "integer"} + + +class BaseIntegerRange(BaseInteger): + _min: int + _max: int + + def _validate(self, context: Context) -> None: + super()._validate(context) + if context.strictness > Strictness.PERMISSIVE and hasattr(self, "_min") and (self._value < self._min): + msg = f"value {self._value} is lower than the minimum {self._min}." + raise DataValueError(msg, self._tree_path) + if context.strictness > Strictness.PERMISSIVE and hasattr(self, "_max") and (self._value > self._max): + msg = f"value {self._value} is higher than the maximum {self._max}" + raise DataValueError(msg, self._tree_path) + + @classmethod + def json_schema(cls) -> dict[Any, Any]: + typ: dict[str, Any] = {"type": "integer"} + if hasattr(cls, "_min"): + typ["minimum"] = cls._min + if hasattr(cls, "_max"): + typ["maximum"] = cls._max + return typ diff --git a/python/knot_resolver/utils/modeling/types/base_path_types.py b/python/knot_resolver/utils/modeling/types/base_path_types.py new file mode 100644 index 000000000..335756a8f --- /dev/null +++ b/python/knot_resolver/utils/modeling/types/base_path_types.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from knot_resolver.logging import get_logger +from knot_resolver.utils.modeling.context import Strictness +from knot_resolver.utils.modeling.errors import DataTypeError + +from .base_custom_type import BaseCustomType + +if TYPE_CHECKING: + from knot_resolver.utils.modeling.context import Context + +logger = get_logger(__name__) + + +class BasePath(BaseCustomType): + """Base class to work with pathlib.Path value.""" + + _path: Path + _path_absolute: Path + + def _validate(self, context: Context) -> None: + if context.strictness > Strictness.PERMISSIVE and not isinstance(self._value, str): + msg = ( + f"Unexpected value for '{type(self)}'" + f" Expected string, got '{self._value}' with type '{type(self._value)}'" + ) + raise DataTypeError(msg) + self._path = Path(self._value) + self._path_absolute = self._path if self._path.is_absolute() else self._base_path / self._path + + @classmethod + def json_schema(cls) -> dict[Any, Any]: + return {"type": "string"} diff --git a/python/knot_resolver/utils/modeling/types/base_string_types.py b/python/knot_resolver/utils/modeling/types/base_string_types.py new file mode 100644 index 000000000..58401e630 --- /dev/null +++ b/python/knot_resolver/utils/modeling/types/base_string_types.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import re +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from knot_resolver.utils.modeling.context import Context, Strictness +from knot_resolver.utils.modeling.errors import DataTypeError, DataValueError + +from .base_custom_type import BaseCustomType + +if TYPE_CHECKING: + from re import Pattern + + +class BaseString(BaseCustomType): + """Base class to work with string value.""" + + def parse(self) -> None: + # parsing is done trough validation with NORMAL strictness + self.validate() + + def _validate(self, context: Context) -> None: + if context.strictness > Strictness.PERMISSIVE and not isinstance(self._value, str): + msg = ( + f"Unexpected value for '{type(self)}'." + f" Expected string, got '{self._value}' with type '{type(self._value)}'" + ) + raise DataTypeError(msg, self._tree_path) + + @classmethod + def json_schema(cls) -> dict[Any, Any]: + return {"type": "string"} + + +class BaseStringLength(BaseString): + _min_bytes: int = 1 + _max_bytes: int + + def _validate(self, context: Context) -> None: + super()._validate(context) + if context.strictness > Strictness.PERMISSIVE: + value_bytes = len(self._value.encode("utf-8")) + if hasattr(self, "_min_bytes") and (value_bytes < self._min_bytes): + msg = f"the string value {self._value} is shorter than the minimum {self._min_bytes} bytes." + raise DataValueError(msg, self._tree_path) + if hasattr(self, "_max_bytes") and (value_bytes > self._max_bytes): + msg = f"the string value {self._value} is longer than the maximum {self._max_bytes} bytes." + raise DataValueError(msg, self._tree_path) + + @classmethod + def json_schema(cls) -> dict[Any, Any]: + typ: dict[str, Any] = {"type": "string"} + if hasattr(cls, "_min_bytes"): + typ["minLength"] = cls._min_bytes + if hasattr(cls, "_max_bytes"): + typ["maxLength"] = cls._max_bytes + return typ + + +class BaseStringPattern(BaseString): + _re: Pattern[str] + + def _validate(self, context: Context) -> None: + super()._validate(context) + if context.strictness > Strictness.PERMISSIVE and not type(self)._re.match(self._value): # noqa: SLF001 + msg = f"'{self._value}' does not match '{self._re.pattern}' pattern" + raise DataValueError(msg, self._tree_path) + + @classmethod + def json_schema(cls) -> dict[Any, Any]: + return {"type": "string", "pattern": rf"{cls._re.pattern}"} + + +class BaseUnit(BaseString): + _re: Pattern[str] + _units: dict[str, int] + _base_value: int | float + + def __init__(self, value: Any, tree_path: str = "/", base_path: Path = Path()) -> None: + super().__init__(value, tree_path, base_path) + type(self)._re = re.compile(rf"^(\d+)({r'|'.join(type(self)._units.keys())})$") # noqa: SLF001 + + def get_base_value(self) -> int | float: + if not self._is_valid: + self.validate() + return self._base_value + + def _validate(self, context: Context) -> None: + super()._validate(context) + + if context.strictness > Strictness.PERMISSIVE: + cls = self.__class__ + grouped = self._re.search(self._value) + + if not grouped: + msg = ( + f"Unexpected value for '{type(self)}'." + " Expected string that matches pattern " + rf"'{type(self)._re.pattern}'." # noqa: SLF001 + f" Positive integer and one of the units {list(type(self)._units.keys())}, got '{self._value}'." # noqa: SLF001 + ) + raise DataValueError(msg, self._tree_path) + value, unit = grouped.groups() + if unit is None: + msg = f"Missing units. Accepted units are {list(cls._units.keys())}" + raise DataValueError(msg, self._tree_path) + if unit not in cls._units: + msg = ( + f"Used unexpected unit '{unit}' for {type(self).__name__}." + f" Accepted units are {list(cls._units.keys())}" + ) + raise DataValueError(msg, self._tree_path) + self._base_value = float(value) * cls._units[unit] + + def __int__(self) -> int: + return int(self.get_base_value()) + + @classmethod + def json_schema(cls) -> dict[Any, Any]: + return {"type": "string", "pattern": rf"{cls._re.pattern}"} diff --git a/tests/python/knot_resolver/utils/modeling/types/test_base_float_types.py b/tests/python/knot_resolver/utils/modeling/types/test_base_float_types.py new file mode 100644 index 000000000..478e3f21d --- /dev/null +++ b/tests/python/knot_resolver/utils/modeling/types/test_base_float_types.py @@ -0,0 +1,79 @@ +import random +import sys +from typing import Any, Optional + +import pytest + +from knot_resolver.utils.modeling.errors import DataModelingError +from knot_resolver.utils.modeling.types.base_float_types import BaseFloat, BaseFloatRange + + +@pytest.mark.parametrize("value", [-65.535, -1, 0, 1, 65.535]) +def test_base_float(value: int): + obj = BaseFloat(value) + obj.validate() + assert float(obj) == value + assert int(obj) == int(value) + assert str(obj) == f"{value}" + + +@pytest.mark.parametrize("value", [True, False, "1"]) +def test_base_float_invalid(value: Any): + with pytest.raises(DataModelingError): + BaseFloat(value).validate() + + +@pytest.mark.parametrize("min,max", [(0.0, None), (None, 0.0), (1.5, 65.535), (-65.535, -1.5)]) +def test_base_float_range(min: Optional[float], max: Optional[float]): + class TestFloatRange(BaseFloatRange): + if min: + _min = min + if max: + _max = max + + if min: + obj = TestFloatRange(min) + obj.validate() + assert float(obj) == min + assert int(obj) == int(min) + assert str(obj) == f"{min}" + if max: + obj = TestFloatRange(max) + obj.validate() + assert float(obj) == max + assert int(obj) == int(max) + assert str(obj) == f"{max}" + + rmin = int(min + 1) if min else -sys.maxsize - 1 + rmax = int(max - 1) if max else sys.maxsize + + n = 100 + values = [float(random.randint(rmin, rmax)) for _ in range(n)] + + for value in values: + obj = TestFloatRange(value) + obj.validate() + assert float(obj) == float(value) + assert str(obj) == f"{value}" + + +@pytest.mark.parametrize("min,max", [(0.0, None), (None, 0.0), (1.5, 65.535), (-65.535, -1.5)]) +def test_base_float_range_invalid(min: Optional[float], max: Optional[float]): + class TestFloatRange(BaseFloatRange): + if min: + _min = min + if max: + _max = max + + n = 100 + invalid_nums = [] + + rmin = int(min + 1) if min else -sys.maxsize - 1 + rmax = int(max - 1) if max else sys.maxsize + + invalid_nums.extend([float(random.randint(rmax + 1, sys.maxsize)) for _ in range(n % 2)] if max else []) + invalid_nums.extend([float(random.randint(-sys.maxsize - 1, rmin - 1)) for _ in range(n % 2)] if max else []) + + for num in invalid_nums: + with pytest.raises(DataModelingError): + TestFloatRange(num).validate() diff --git a/tests/python/knot_resolver/utils/modeling/types/test_base_generic_custom_types.py b/tests/python/knot_resolver/utils/modeling/types/test_base_generic_custom_types.py new file mode 100644 index 000000000..327339c52 --- /dev/null +++ b/tests/python/knot_resolver/utils/modeling/types/test_base_generic_custom_types.py @@ -0,0 +1,37 @@ +from typing import Any, List, Union + +import pytest + +from knot_resolver.utils.modeling.types.base_generic_custom_types import ListOrItem, Transformed +from knot_resolver.utils.modeling.types.inspect import ( + get_base_generic_type_wrapper_argument, + get_transformed_input_type, + get_transformed_result_type, +) + + +@pytest.mark.parametrize("result_t,input_t", [(str, int), (float, bool)]) +def test_transformed_inner(result_t: Any, input_t: Any) -> None: + typ = Transformed[result_t, input_t] + assert get_transformed_input_type(typ) == input_t + assert get_transformed_result_type(typ) == result_t + + +@pytest.mark.parametrize("typ", [str, int, float, bool]) +def test_list_or_item_inner_type(typ: Any) -> None: + assert get_base_generic_type_wrapper_argument(ListOrItem[typ]) == Union[List[typ], typ] + + +@pytest.mark.parametrize( + "value", + [ + [], + 65_535, + [1, 65_535, 5335, 5000], + ], +) +def test_list_or_item(value: Any) -> None: + obj = ListOrItem(value) + assert str(obj) == str(value) + for i, item in enumerate(obj): + assert item == value[i] if isinstance(value, list) else value diff --git a/tests/python/knot_resolver/utils/modeling/types/test_base_integer_types.py b/tests/python/knot_resolver/utils/modeling/types/test_base_integer_types.py new file mode 100644 index 000000000..9e88041fb --- /dev/null +++ b/tests/python/knot_resolver/utils/modeling/types/test_base_integer_types.py @@ -0,0 +1,75 @@ +import random +import sys +from typing import Any, Optional + +import pytest + +from knot_resolver.utils.modeling.errors import DataModelingError +from knot_resolver.utils.modeling.types.base_integer_types import BaseInteger, BaseIntegerRange + + +@pytest.mark.parametrize("value", [-65535, -1, 0, 1, 65535]) +def test_base_integer(value: int): + obj = BaseInteger(value) + obj.validate() + assert int(obj) == value + assert str(obj) == f"{value}" + + +@pytest.mark.parametrize("value", [True, False, "1", 1.1]) +def test_base_integer_invalid(value: Any): + with pytest.raises(DataModelingError): + BaseInteger(value).validate() + + +@pytest.mark.parametrize("min,max", [(0, None), (None, 0), (1, 65535), (-65535, -1)]) +def test_base_integer_range(min: Optional[int], max: Optional[int]): + class TestIntegerRange(BaseIntegerRange): + if min: + _min = min + if max: + _max = max + + if min: + obj = TestIntegerRange(min) + obj.validate() + assert int(obj) == min + assert str(obj) == f"{min}" + if max: + obj = TestIntegerRange(max) + obj.validate() + assert int(obj) == max + assert str(obj) == f"{max}" + + rmin = min if min else -sys.maxsize - 1 + rmax = max if max else sys.maxsize + + n = 100 + values = [random.randint(rmin, rmax) for _ in range(n)] + + for value in values: + obj = TestIntegerRange(value) + obj.validate() + assert str(obj) == f"{value}" + + +@pytest.mark.parametrize("min,max", [(0, None), (None, 0), (1, 65535), (-65535, -1)]) +def test_base_integer_range_invalid(min: Optional[int], max: Optional[int]): + class TestIntegerRange(BaseIntegerRange): + if min: + _min = min + if max: + _max = max + + n = 100 + invalid_nums = [] + + rmin = min if min else -sys.maxsize - 1 + rmax = max if max else sys.maxsize + + invalid_nums.extend([random.randint(rmax + 1, sys.maxsize) for _ in range(n % 2)] if max else []) + invalid_nums.extend([random.randint(-sys.maxsize - 1, rmin - 1) for _ in range(n % 2)] if max else []) + + for num in invalid_nums: + with pytest.raises(DataModelingError): + TestIntegerRange(num).validate() diff --git a/tests/python/knot_resolver/utils/modeling/types/test_base_path_types.py b/tests/python/knot_resolver/utils/modeling/types/test_base_path_types.py new file mode 100644 index 000000000..91f3006ce --- /dev/null +++ b/tests/python/knot_resolver/utils/modeling/types/test_base_path_types.py @@ -0,0 +1,34 @@ +from pathlib import Path +from typing import Any + +import pytest + +from knot_resolver.utils.modeling.context import Context, Strictness +from knot_resolver.utils.modeling.errors import DataModelingError +from knot_resolver.utils.modeling.types.base_path_types import BasePath + +context_default = Context(strictness=Strictness.BASIC) +base_path = Path("/base/path/prefix") + + +@pytest.mark.parametrize( + "value", + [ + "relative/path/to/dir", + "relative/path/to/file.txt", + "/absolute/path/to/dir", + "/absolute/path/to/file.txt", + ], +) +def test_base_path(value: str): + obj = BasePath(value, base_path=base_path) + obj.validate(context_default) + assert obj._path == Path(value) + assert obj._path_absolute == Path(value) if value.startswith("/") else base_path / value + + +@pytest.mark.parametrize("value", [1, 1.1, True, False]) +def test_base_path_invalid(value: Any): + obj = BasePath(value, base_path=base_path) + with pytest.raises(DataModelingError): + obj.validate(context_default) diff --git a/tests/python/knot_resolver/utils/modeling/types/test_base_string_types.py b/tests/python/knot_resolver/utils/modeling/types/test_base_string_types.py new file mode 100644 index 000000000..05c756616 --- /dev/null +++ b/tests/python/knot_resolver/utils/modeling/types/test_base_string_types.py @@ -0,0 +1,109 @@ +import random +import string +from typing import Any, Optional + +import pytest + +from knot_resolver.utils.modeling.errors import DataModelingError +from knot_resolver.utils.modeling.types.base_string_types import BaseString, BaseStringLength, BaseUnit + + +@pytest.mark.parametrize("value", ["a", "abcdef"]) +def test_base_string(value: str): + obj = BaseString(value) + obj.validate() + assert str(obj) == str(value) + + +@pytest.mark.parametrize("value", [1234, True, False]) +def test_base_string_invalid(value: Any): + with pytest.raises(DataModelingError): + BaseString(value).validate() + + +@pytest.mark.parametrize("min,max", [(None, 100), (10, 20), (50, None)]) +def test_base_string_length(min: Optional[int], max: Optional[int]): + class TestStringLength(BaseStringLength): + if min: + _min_bytes = min + if max: + _max_bytes = max + + if min: + rand_str = "".join(random.choices(string.ascii_uppercase + string.digits, k=min)) + obj = TestStringLength(rand_str) + obj.validate() + assert str(obj) == f"{rand_str}" + if max: + rand_str = "".join(random.choices(string.ascii_uppercase + string.digits, k=max)) + obj = TestStringLength(rand_str) + obj.validate() + assert str(obj) == f"{rand_str}" + + rmin = min if min else 1 + rmax = max if max else 200 + + n = 100 + values = [ + "".join(random.choices(string.ascii_uppercase + string.digits, k=random.randint(rmin, rmax))) for _ in range(n) + ] + + for value in values: + obj = TestStringLength(value) + obj.validate() + assert str(obj) == f"{value}" + + +@pytest.mark.parametrize("min,max", [(None, 100), (10, 20), (50, None)]) +def test_base_string_length_invalid(min: Optional[int], max: Optional[int]): + class TestStringLength(BaseStringLength): + if min: + _min_bytes = min + if max: + _max_bytes = max + + n = 100 + invalid_strings = [] + + rmin = min if min else 1 + rmax = max if max else 200 + + invalid_strings.extend( + [ + "".join(random.choices(string.ascii_uppercase + string.digits, k=random.randint(rmax, rmax + 20))) + for _ in range(n % 2) + ] + if max + else [] + ) + invalid_strings.extend( + [ + "".join(random.choices(string.ascii_uppercase + string.digits, k=random.randint(1, rmin))) + for _ in range(n % 2) + ] + if max + else [] + ) + + for invalid_string in invalid_strings: + with pytest.raises(DataModelingError): + TestStringLength(invalid_string).validate() + + +@pytest.mark.parametrize("value", ["1000a", "100b", "10c", "1d"]) +def test_base_unit(value: str): + class TestBaseUnit(BaseUnit): + _units = {"a": 1, "b": 10, "c": 100, "d": 1000} + + obj = TestBaseUnit(value) + obj.validate() + assert int(obj) == 1000 + + +@pytest.mark.parametrize("value", [True, False, "1000aa", "10ab", "1e"]) +def test_base_unit_invalid(value: Any): + class TestBaseUnit(BaseUnit): + _units = {"a": 1, "b": 10, "c": 100, "d": 1000} + + with pytest.raises(DataModelingError): + TestBaseUnit(value).validate()