manager: datamodel: custom type for IP address with '!' suffix

diff --git a/manager/knot_resolver_manager/datamodel/network_schema.py b/manager/knot_resolver_manager/datamodel/network_schema.py
index aacb2c4692af3e7a3c184c1587494f8233b6a93a..e59fe4137e1d71dee836bcc7197e57e58af111d7 100644 (file)
--- a/manager/knot_resolver_manager/datamodel/network_schema.py
+++ b/manager/knot_resolver_manager/datamodel/network_schema.py
@@ -10,6 +10,7 @@ from knot_resolver_manager.datamodel.types import (
     Int0_65535,
     InterfaceOptionalPort,
     IPAddress,
+    IPAddressEM,
     IPNetwork,
     IPv4Address,
     IPv6Address,
@@ -45,7 +46,7 @@ class AddressRenumberingSchema(ConfigSchema):
     """
 
     source: IPNetwork
-    destination: IPAddress
+    destination: Union[IPAddressEM, IPAddress]
 
 
 class TLSSchema(ConfigSchema):

diff --git a/manager/knot_resolver_manager/datamodel/types/__init__.py b/manager/knot_resolver_manager/datamodel/types/__init__.py
index c100472bec8dd4996b0278c890c9867545bfd779..350cf2133888e9aa47879b70a1cb86e2d74c952c 100644 (file)
--- a/manager/knot_resolver_manager/datamodel/types/__init__.py
+++ b/manager/knot_resolver_manager/datamodel/types/__init__.py
@@ -14,6 +14,7 @@ from .types import (
     IntNonNegative,
     IntPositive,
     IPAddress,
+    IPAddressEM,
     IPAddressOptionalPort,
     IPAddressPort,
     IPNetwork,
@@ -44,6 +45,7 @@ __all__ = [
     "IntNonNegative",
     "IntPositive",
     "IPAddress",
+    "IPAddressEM",
     "IPAddressOptionalPort",
     "IPAddressPort",
     "IPNetwork",

diff --git a/manager/knot_resolver_manager/datamodel/types/types.py b/manager/knot_resolver_manager/datamodel/types/types.py
index f558da0de24135de9ef9ede81c033ce974383b30..19bc75bb20b382db56574f386962e8e3b9fa1da0 100644 (file)
--- a/manager/knot_resolver_manager/datamodel/types/types.py
+++ b/manager/knot_resolver_manager/datamodel/types/types.py
@@ -376,6 +376,63 @@ class IPv6Address(BaseValueType):
 IPAddress = Union[IPv4Address, IPv6Address]
 
 
+class IPAddressEM(BaseValueType):
+    """
+    IP address with exclamation mark suffix, e.g. '127.0.0.1!'.
+    """
+
+    _value: str
+    _addr: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
+
+    def __init__(self, source_value: Any, object_path: str = "/") -> None:
+        super().__init__(source_value)
+        if isinstance(source_value, str):
+            if source_value.endswith("!"):
+                addr, suff = source_value.split("!", 1)
+                if suff != "":
+                    raise ValueError(f"suffix '{suff}' found after '!'.")
+            else:
+                raise ValueError("string does not end with '!'.")
+            try:
+                self._addr: Union[ipaddress.IPv4Address, ipaddress.IPv6Address] = ipaddress.ip_address(addr)
+                self._value = source_value
+            except ValueError as e:
+                raise ValueError("failed to parse IP address.") from e
+        else:
+            raise ValueError(
+                "Unexpected value for a IPv6 address."
+                f" Expected string, got '{source_value}' with type '{type(source_value)}'",
+                object_path,
+            )
+
+    def to_std(self) -> str:
+        return self._value
+
+    def __str__(self) -> str:
+        return self._value
+
+    def __int__(self) -> int:
+        raise ValueError("Can't convert to an integer")
+
+    def __repr__(self) -> str:
+        return f'{type(self).__name__}("{self._value}")'
+
+    def __eq__(self, o: object) -> bool:
+        """
+        Two instances of IPAddressEM are equal when they represent same string.
+        """
+        return isinstance(o, IPAddressEM) and o._value == self._value
+
+    def serialize(self) -> Any:
+        return self._value
+
+    @classmethod
+    def json_schema(cls: Type["IPAddressEM"]) -> Dict[Any, Any]:
+        return {
+            "type": "string",
+        }
+
+
 class IPNetwork(BaseValueType):
     _value: Union[ipaddress.IPv4Network, ipaddress.IPv6Network]