]> git.ipfire.org Git - thirdparty/knot-resolver.git/commitdiff
manager: utils: modeling: raise ValueError instead of DataValidationError in custom...
authorAleš Mrázek <ales.mrazek@nic.cz>
Wed, 13 Jul 2022 09:29:09 +0000 (11:29 +0200)
committerAleš Mrázek <ales.mrazek@nic.cz>
Wed, 13 Jul 2022 09:29:09 +0000 (11:29 +0200)
manager/knot_resolver_manager/datamodel/types/base_types.py
manager/knot_resolver_manager/datamodel/types/types.py
manager/knot_resolver_manager/utils/modeling/base_custom_type.py
manager/knot_resolver_manager/utils/modeling/base_schema.py
manager/tests/unit/datamodel/types/test_custom_types.py

index 274fd2eb37f6c4942dd23b7f48b0df10decca6d6..d6830c83af17c45b9442d9ab980703386b05cea2 100644 (file)
@@ -2,7 +2,6 @@ import re
 from typing import Any, Dict, Pattern, Type
 
 from knot_resolver_manager.utils.modeling import BaseCustomType
-from knot_resolver_manager.utils.modeling.exceptions import DataValidationError
 
 
 class IntBase(BaseCustomType):
@@ -75,12 +74,12 @@ class IntRangeBase(IntBase):
         super().__init__(source_value)
         if isinstance(source_value, int) and not isinstance(source_value, bool):
             if hasattr(self, "_min") and (source_value < self._min):
-                raise DataValidationError(f"value {source_value} is lower than the minimum {self._min}.", object_path)
+                raise ValueError(f"value {source_value} is lower than the minimum {self._min}.")
             if hasattr(self, "_max") and (source_value > self._max):
-                raise DataValidationError(f"value {source_value} is higher than the maximum {self._max}", object_path)
+                raise ValueError(f"value {source_value} is higher than the maximum {self._max}")
             self._value = source_value
         else:
-            raise DataValidationError(
+            raise ValueError(
                 f"expected integer, got '{type(source_value)}'",
                 object_path,
             )
@@ -112,9 +111,9 @@ class PatternBase(StrBase):
             if type(self)._re.match(source_value):
                 self._value: str = source_value
             else:
-                raise DataValidationError(f"'{source_value}' does not match '{self._re.pattern}' pattern", object_path)
+                raise ValueError(f"'{source_value}' does not match '{self._re.pattern}' pattern")
         else:
-            raise DataValidationError(
+            raise ValueError(
                 f"expected string, got '{type(source_value)}'",
                 object_path,
             )
@@ -146,25 +145,23 @@ class UnitBase(IntBase):
             if grouped:
                 val, unit = grouped.groups()
                 if unit is None:
-                    raise DataValidationError(
-                        f"Missing units. Accepted units are {list(type(self)._units.keys())}", object_path
-                    )
+                    raise ValueError(f"Missing units. Accepted units are {list(type(self)._units.keys())}")
                 elif unit not in type(self)._units:
-                    raise DataValidationError(
+                    raise ValueError(
                         f"Used unexpected unit '{unit}' for {type(self).__name__}."
                         f" Accepted units are {list(type(self)._units.keys())}",
                         object_path,
                     )
                 self._value = int(val) * type(self)._units[unit]
             else:
-                raise DataValidationError(f"{type(self._value)} Failed to convert: {self}", object_path)
+                raise ValueError(f"{type(self._value)} Failed to convert: {self}")
         elif isinstance(source_value, int):
-            raise DataValidationError(
+            raise ValueError(
                 f"number without units, please convert to string and add unit  - {list(type(self)._units.keys())}",
                 object_path,
             )
         else:
-            raise DataValidationError(
+            raise ValueError(
                 f"expected number with units in a string, got '{type(source_value)}'.",
                 object_path,
             )
index bb6e635b138d7ea4f219aabc94be59330b6911a9..9d00d175413910b1ee3b2568614f7c61af69703b 100644 (file)
@@ -5,7 +5,6 @@ from typing import Any, Dict, Optional, Type, Union
 
 from knot_resolver_manager.datamodel.types.base_types import IntRangeBase, PatternBase, StrBase, UnitBase
 from knot_resolver_manager.utils.modeling import BaseCustomType
-from knot_resolver_manager.utils.modeling.exceptions import DataValidationError
 
 
 class IntNonNegative(IntRangeBase):
@@ -35,7 +34,7 @@ class PortNumber(IntRangeBase):
         try:
             return cls(int(port), object_path)
         except ValueError as e:
-            raise DataValidationError(f"invalid port number {port}", object_path) from e
+            raise ValueError(f"invalid port number {port}") from e
 
 
 class SizeUnit(UnitBase):
@@ -77,7 +76,7 @@ class DomainName(StrBase):
             try:
                 punycode = source_value.encode("idna").decode("utf-8") if source_value != "." else "."
             except ValueError:
-                raise DataValidationError(
+                raise ValueError(
                     f"conversion of '{source_value}' to IDN punycode representation failed",
                     object_path,
                 )
@@ -86,12 +85,12 @@ class DomainName(StrBase):
                 self._value = source_value
                 self._punycode = punycode
             else:
-                raise DataValidationError(
+                raise ValueError(
                     f"'{source_value}' represented in punycode '{punycode}' does not match '{self._re.pattern}' pattern",
                     object_path,
                 )
         else:
-            raise DataValidationError(
+            raise ValueError(
                 "Unexpected value for '<domain-name>'."
                 f" Expected string, got '{source_value}' with type '{type(source_value)}'",
                 object_path,
@@ -137,18 +136,14 @@ class InterfacePort(StrBase):
                 except ValueError as e1:
                     try:
                         self.if_name = InterfaceName(parts[0])
-                    except DataValidationError as e2:
-                        raise DataValidationError(
-                            f"expected IP address or interface name, got '{parts[0]}'.", object_path
-                        ) from e1 and e2
+                    except ValueError as e2:
+                        raise ValueError(f"expected IP address or interface name, got '{parts[0]}'.") from e1 and e2
                 self.port = PortNumber.from_str(parts[1], object_path)
             else:
-                raise DataValidationError(
-                    f"expected '<ip-address|interface-name>@<port>', got '{source_value}'.", object_path
-                )
+                raise ValueError(f"expected '<ip-address|interface-name>@<port>', got '{source_value}'.")
             self._value = source_value
         else:
-            raise DataValidationError(
+            raise ValueError(
                 "Unexpected value for '<ip-address|interface-name>@<port>'."
                 f" Expected string, got '{source_value}' with type '{type(source_value)}'",
                 object_path,
@@ -170,19 +165,15 @@ class InterfaceOptionalPort(StrBase):
                 except ValueError as e1:
                     try:
                         self.if_name = InterfaceName(parts[0])
-                    except DataValidationError as e2:
-                        raise DataValidationError(
-                            f"expected IP address or interface name, got '{parts[0]}'.", object_path
-                        ) from e1 and e2
+                    except ValueError as e2:
+                        raise ValueError(f"expected IP address or interface name, got '{parts[0]}'.") from e1 and e2
                 if len(parts) == 2:
                     self.port = PortNumber.from_str(parts[1], object_path)
             else:
-                raise DataValidationError(
-                    f"expected '<ip-address|interface-name>[@<port>]', got '{parts}'.", object_path
-                )
+                raise ValueError(f"expected '<ip-address|interface-name>[@<port>]', got '{parts}'.")
             self._value = source_value
         else:
-            raise DataValidationError(
+            raise ValueError(
                 "Unexpected value for '<ip-address|interface-name>[@<port>]'."
                 f" Expected string, got '{source_value}' with type '{type(source_value)}'",
                 object_path,
@@ -202,15 +193,14 @@ class IPAddressPort(StrBase):
                 try:
                     self.addr = ipaddress.ip_address(parts[0])
                 except ValueError as e:
-                    raise DataValidationError(f"failed to parse IP address '{parts[0]}'.", object_path) from e
+                    raise ValueError(f"failed to parse IP address '{parts[0]}'.") from e
             else:
-                raise DataValidationError(f"expected '<ip-address>@<port>', got '{source_value}'.", object_path)
+                raise ValueError(f"expected '<ip-address>@<port>', got '{source_value}'.")
             self._value = source_value
         else:
-            raise DataValidationError(
+            raise ValueError(
                 "Unexpected value for '<ip-address>@<port>'."
-                f" Expected string, got '{source_value}' with type '{type(source_value)}'",
-                object_path,
+                f" Expected string, got '{source_value}' with type '{type(source_value)}'"
             )
 
 
@@ -226,14 +216,14 @@ class IPAddressOptionalPort(StrBase):
                 try:
                     self.addr = ipaddress.ip_address(parts[0])
                 except ValueError as e:
-                    raise DataValidationError(f"failed to parse IP address '{parts[0]}'.", object_path) from e
+                    raise ValueError(f"failed to parse IP address '{parts[0]}'.") from e
                 if len(parts) == 2:
                     self.port = PortNumber.from_str(parts[1], object_path)
             else:
-                raise DataValidationError(f"expected '<ip-address>[@<port>]', got '{parts}'.", object_path)
+                raise ValueError(f"expected '<ip-address>[@<port>]', got '{parts}'.")
             self._value = source_value
         else:
-            raise DataValidationError(
+            raise ValueError(
                 "Unexpected value for a '<ip-address>[@<port>]'."
                 f" Expected string, got '{source_value}' with type '{type(source_value)}'",
                 object_path,
@@ -249,9 +239,9 @@ class IPv4Address(BaseCustomType):
             try:
                 self._value: ipaddress.IPv4Address = ipaddress.IPv4Address(source_value)
             except ValueError as e:
-                raise DataValidationError("failed to parse IPv4 address.", object_path) from e
+                raise ValueError("failed to parse IPv4 address.") from e
         else:
-            raise DataValidationError(
+            raise ValueError(
                 "Unexpected value for a IPv4 address."
                 f" Expected string, got '{source_value}' with type '{type(source_value)}'",
                 object_path,
@@ -291,9 +281,9 @@ class IPv6Address(BaseCustomType):
             try:
                 self._value: ipaddress.IPv6Address = ipaddress.IPv6Address(source_value)
             except ValueError as e:
-                raise DataValidationError("failed to parse IPv6 address.", object_path) from e
+                raise ValueError("failed to parse IPv6 address.") from e
         else:
-            raise DataValidationError(
+            raise ValueError(
                 "Unexpected value for a IPv6 address."
                 f" Expected string, got '{source_value}' with type '{type(source_value)}'",
                 object_path,
@@ -336,12 +326,11 @@ class IPNetwork(BaseCustomType):
             try:
                 self._value: Union[ipaddress.IPv4Network, ipaddress.IPv6Network] = ipaddress.ip_network(source_value)
             except ValueError as e:
-                raise DataValidationError("failed to parse IP network.", object_path) from e
+                raise ValueError("failed to parse IP network.") from e
         else:
-            raise DataValidationError(
+            raise ValueError(
                 "Unexpected value for a network subnet."
-                f" Expected string, got '{source_value}' with type '{type(source_value)}'",
-                object_path,
+                f" Expected string, got '{source_value}' with type '{type(source_value)}'"
             )
 
     def to_std(self) -> Union[ipaddress.IPv4Network, ipaddress.IPv6Network]:
@@ -372,27 +361,24 @@ class IPv6Network96(BaseCustomType):
             try:
                 self._value: ipaddress.IPv6Network = ipaddress.IPv6Network(source_value)
             except ValueError as e:
-                raise DataValidationError("failed to parse IPv6 /96 network.", object_path) from e
+                raise ValueError("failed to parse IPv6 /96 network.") from e
 
             if self._value.prefixlen == 128:
-                raise DataValidationError(
+                raise ValueError(
                     "Expected IPv6 network address with /96 prefix length."
                     " Submitted address has been interpreted as /128."
-                    " Maybe, you forgot to add /96 after the base address?",
-                    object_path,
+                    " Maybe, you forgot to add /96 after the base address?"
                 )
 
             if self._value.prefixlen != 96:
-                raise DataValidationError(
+                raise ValueError(
                     "expected IPv6 network address with /96 prefix length."
-                    f" Got prefix lenght of {self._value.prefixlen}",
-                    object_path,
+                    f" Got prefix lenght of {self._value.prefixlen}"
                 )
         else:
-            raise DataValidationError(
+            raise ValueError(
                 "Unexpected value for a network subnet."
-                f" Expected string, got '{source_value}' with type '{type(source_value)}'",
-                object_path,
+                f" Expected string, got '{source_value}' with type '{type(source_value)}'"
             )
 
     def __str__(self) -> str:
@@ -428,9 +414,7 @@ class UncheckedPath(BaseCustomType):
         if isinstance(source_value, str):
             self._value: Path = Path(source_value)
         else:
-            raise DataValidationError(
-                f"expected file path in a string, got '{source_value}' with type '{type(source_value)}'.", object_path
-            )
+            raise ValueError(f"expected file path in a string, got '{source_value}' with type '{type(source_value)}'.")
 
     def __str__(self) -> str:
         return str(self._value)
@@ -468,4 +452,4 @@ class CheckedPath(UncheckedPath):
         try:
             self._value = self._value.resolve(strict=False)
         except RuntimeError as e:
-            raise DataValidationError("Failed to resolve given file path. Is there a symlink loop?", object_path) from e
+            raise ValueError("Failed to resolve given file path. Is there a symlink loop?") from e
index 1da6fe812afa2dc7e287fe7b46a15cde38176c6b..ccde09992e315666c3e2dcc4ab361673df478775 100644 (file)
@@ -18,7 +18,7 @@ class BaseCustomType:
 
     There is no validation done on the wrapped value. The only condition is that
     it can't be `None`. If you want to perform any validation during creation,
-    raise a `DataValidationError` in case of errors.
+    raise a `ValueError` in case of errors.
     """
 
     def __init__(self, source_value: Any, object_path: str = "/") -> None:
index 09227e1e434386ee2dd9bf60d7bca2d1d26f7645..9278c2fcef4a221063b15593a87408925e1ea04f 100644 (file)
@@ -365,7 +365,14 @@ def _validated_object_type(
             return obj
         else:
             # no validation performed, the implementation does it in the constuctor
-            return cls(obj, object_path=object_path)
+            try:
+                return cls(obj, object_path=object_path)
+            except ValueError as e:
+                if len(e.args) > 0 and isinstance(e.args[0], str):
+                    msg = e.args[0]
+                else:
+                    msg = f"Failed to validate value against {cls} type"
+                raise DataValidationError(msg, object_path) from e
 
     # nested BaseSchema subclasses
     elif inspect.isclass(cls) and issubclass(cls, BaseSchema):
index b63520bd0beb08ca481559f685d9648251659dd8..2f9b1c1928317272079d3b05f944e3a5ef772e47 100644 (file)
@@ -23,7 +23,6 @@ from knot_resolver_manager.datamodel.types import (
     TimeUnit,
 )
 from knot_resolver_manager.utils.modeling import BaseSchema
-from knot_resolver_manager.utils.modeling.exceptions import DataValidationError
 
 
 def _rand_domain(label_chars: int, levels: int = 1) -> str:
@@ -39,7 +38,7 @@ def test_port_number_valid(val: int):
 
 @pytest.mark.parametrize("val", [0, 65_636, -1, "53"])
 def test_port_number_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         PortNumber(val)
 
 
@@ -53,7 +52,7 @@ def test_size_unit_valid(val: str):
 
 @pytest.mark.parametrize("val", ["-5B", 5, -5242880, "45745mB"])
 def test_size_unit_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         SizeUnit(val)
 
 
@@ -68,7 +67,7 @@ def test_time_unit_valid(val: str):
 
 @pytest.mark.parametrize("val", ["-1", "-24h", "1440mm", 6575, -1440])
 def test_time_unit_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         TimeUnit("-1")
 
 
@@ -124,7 +123,7 @@ def test_domain_name_valid(val: str):
     ],
 )
 def test_domain_name_invalid(val: str):
-    with raises(DataValidationError):
+    with raises(ValueError):
         DomainName(val)
 
 
@@ -135,7 +134,7 @@ def test_interface_name_valid(val: str):
 
 @pytest.mark.parametrize("val", ["_lo", "-wlo1", "lo_", "wlo1-", "e8--2", "web__ifgrp"])
 def test_interface_name_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         InterfaceName(val)
 
 
@@ -150,7 +149,7 @@ def test_interface_port_valid(val: str):
 
 @pytest.mark.parametrize("val", ["lo", "2001:db8::1000", "53"])
 def test_interface_port_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         InterfacePort(val)
 
 
@@ -165,7 +164,7 @@ def test_interface_optional_port_valid(val: str):
 
 @pytest.mark.parametrize("val", ["lo@", "@53"])
 def test_interface_optional_port_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         InterfaceOptionalPort(val)
 
 
@@ -182,7 +181,7 @@ def test_ip_address_port_valid(val: str):
     "val", ["123.4.5.6", "2001:db8::1000", "123.4.5.6.7@5000", "2001:db8::10000@5001", "123.4.5.6@"]
 )
 def test_ip_address_port_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         IPAddressPort(val)
 
 
@@ -197,7 +196,7 @@ def test_ip_address_optional_port_valid(val: str):
 
 @pytest.mark.parametrize("val", ["123.4.5.6.7", "2001:db8::10000", "123.4.5.6@", "@55"])
 def test_ip_address_optional_port_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         IPAddressOptionalPort(val)
 
 
@@ -210,7 +209,7 @@ def test_ipv4_address_valid(val: str):
 
 @pytest.mark.parametrize("val", ["123456", "2001:db8::1000"])
 def test_ipv4_address_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         IPv4Address(val)
 
 
@@ -223,7 +222,7 @@ def test_ipv6_address_valid(val: str):
 
 @pytest.mark.parametrize("val", ["123.4.5.6", "2001::db8::1000"])
 def test_ipv6_address_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         IPv6Address(val)
 
 
@@ -237,7 +236,7 @@ def test_ip_network_valid(val: str):
 
 @pytest.mark.parametrize("val", ["10.11.12.13/8", "10.11.12.5/128"])
 def test_ip_network_invalid(val: str):
-    with raises(DataValidationError):
+    with raises(ValueError):
         IPNetwork(val)
 
 
@@ -248,5 +247,5 @@ def test_ipv6_96_network_valid(val: str):
 
 @pytest.mark.parametrize("val", ["fe80::/95", "10.11.12.3/96", "64:ff9b::1/96"])
 def test_ipv6_96_network_invalid(val: Any):
-    with raises(DataValidationError):
+    with raises(ValueError):
         IPv6Network96(val)