--- /dev/null
+from .base_generic_custom_types import ListOrItem, Transformed
+
+__all__ = [
+ "ListOrItem",
+ "Transformed",
+]
--- /dev/null
+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
--- /dev/null
+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())
--- /dev/null
+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
--- /dev/null
+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"}
--- /dev/null
+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}"}
--- /dev/null
+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()
--- /dev/null
+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
--- /dev/null
+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()
--- /dev/null
+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)
--- /dev/null
+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()
projects / thirdparty / knot-resolver.git / commitdiff
