datamodel: types: added new custom types

- DomainName
- IPAddressPort

diff --git a/manager/knot_resolver_manager/datamodel/types.py b/manager/knot_resolver_manager/datamodel/types.py
index 9e9975753336d48d1fef1710fa89d6c6a3a0f0e4..9a222082b777b8fa4ced6c0ee3bba7cf8326a044 100644 (file)
--- a/manager/knot_resolver_manager/datamodel/types.py
+++ b/manager/knot_resolver_manager/datamodel/types.py
@@ -134,6 +134,106 @@ class AnyPath(CustomValueType):
         }
 
 
+class DomainName(CustomValueType):
+    _re = re.compile(
+        r"^(([a-zA-Z]{1})|([a-zA-Z]{1}[a-zA-Z]{1})|"
+        r"([a-zA-Z]{1}[0-9]{1})|([0-9]{1}[a-zA-Z]{1})|"
+        r"([a-zA-Z0-9][-_.a-zA-Z0-9]{0,61}[a-zA-Z0-9]))\."
+        r"([a-zA-Z]{2,13}|[a-zA-Z0-9-]{2,30}.[a-zA-Z]{2,3})($|.$)"
+    )
+
+    def __init__(self, source_value: Any, object_path: str = "/") -> None:
+        super().__init__(source_value)
+        if isinstance(source_value, str):
+            if type(self)._re.match(source_value):
+                self._value: str = source_value
+            else:
+                raise SchemaException(f"'{source_value}' is not valid domain name", object_path)
+        else:
+            raise SchemaException(
+                f"Unexpected input type for DomainName type - {type(source_value)}."
+                "Cause might be invalid format or invalid type.",
+                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 DomainName to an integer")
+
+    def __eq__(self, o: object) -> bool:
+        """
+        Two instances of DomainName are equal when they represent same string.
+        """
+        return isinstance(o, DomainName) and str(o._value) == str(self._value)
+
+    def serialize(self) -> Any:
+        return str(self._value)
+
+    @classmethod
+    def json_schema(cls: Type["DomainName"]) -> Dict[Any, Any]:
+        return {
+            "type": "string",
+        }
+
+
+class IPAddressPort(CustomValueType):
+    def __init__(self, source_value: Any, object_path: str = "/") -> None:
+        super().__init__(source_value)
+        if isinstance(source_value, str):
+            addr = source_value
+            if "@" in source_value:
+                sep = source_value.split("@", 1)
+                addr = sep[0]
+                try:
+                    port = int(sep[1])
+                except ValueError as e:
+                    raise SchemaException("Failed to parse port.", object_path) from e
+                if not 0 <= port <= 65_535:
+                    raise SchemaException(f"Port value '{port}' out of range of usual 2-byte port value", object_path)
+
+            try:
+                ipaddress.ip_address(addr)
+            except ValueError as e:
+                raise SchemaException("Failed to parse IP address.", object_path) from e
+
+            self._value: str = source_value
+        else:
+            raise SchemaException(
+                f"Unexpected value for a '<ip-address>@<port>'. Expected string, got '{source_value}'"
+                f" 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 IP address to an integer")
+
+    def __eq__(self, o: object) -> bool:
+        """
+        Two instances of IPAddressPORT are equal when they represent same string.
+        """
+        return isinstance(o, IPAddressPort) and str(o._value) == str(self._value)
+
+    def serialize(self) -> Any:
+        return str(self._value)
+
+    @classmethod
+    def json_schema(cls: Type["IPAddressPort"]) -> Dict[Any, Any]:
+        return {
+            "type": "string",
+        }
+
+
 class IPv4Address(CustomValueType):
     def __init__(self, source_value: Any, object_path: str = "/") -> None:
         super().__init__(source_value)

diff --git a/manager/tests/datamodel/test_datamodel_types.py b/manager/tests/datamodel/test_datamodel_types.py
index 4b329776263247a62731a904bd53d8a8bdc249b2..7e9416459c118f22345f43926a641f02070e0641 100644 (file)
--- a/manager/tests/datamodel/test_datamodel_types.py
+++ b/manager/tests/datamodel/test_datamodel_types.py
@@ -4,7 +4,9 @@ from pytest import raises
 
 from knot_resolver_manager.datamodel.types import (
     AnyPath,
+    DomainName,
     IPAddress,
+    IPAddressPort,
     IPv4Address,
     IPv6Address,
     IPNetwork,
@@ -61,6 +63,54 @@ def test_anypath():
     assert str(TestSchema({"p": "/tmp"}).p) == "/tmp"
 
 
+def test_domain_name():
+    class TestSchema(SchemaNode):
+        name: DomainName
+
+    o = TestSchema({"name": "test.domain.com."})
+    assert str(o.name) == "test.domain.com."
+    assert o.name == DomainName("test.domain.com.")
+
+    o = TestSchema({"name": "test.domain.com"})
+    assert str(o.name) == "test.domain.com"
+    assert o.name == DomainName("test.domain.com")
+
+    with raises(KresdManagerException):
+        TestSchema({"name": "b@d.domain.com."})
+
+
+def test_ipaddress_port():
+    class TestSchema(SchemaNode):
+        ip_port: IPAddressPort
+
+    o = TestSchema({"ip-port": "123.4.5.6"})
+    assert str(o.ip_port) == "123.4.5.6"
+    assert o.ip_port == IPAddressPort("123.4.5.6")
+
+    o = TestSchema({"ip-port": "123.4.5.6@5353"})
+    assert str(o.ip_port) == "123.4.5.6@5353"
+    assert o.ip_port == IPAddressPort("123.4.5.6@5353")
+
+    o = TestSchema({"ip-port": "2001:db8::1000"})
+    assert str(o.ip_port) == "2001:db8::1000"
+    assert o.ip_port == IPAddressPort("2001:db8::1000")
+
+    o = TestSchema({"ip-port": "2001:db8::1000@53"})
+    assert str(o.ip_port) == "2001:db8::1000@53"
+    assert o.ip_port == IPAddressPort("2001:db8::1000@53")
+
+    with raises(KresdManagerException):
+        TestSchema({"ip-port": "123.4.5.6.7"})
+    with raises(KresdManagerException):
+        TestSchema({"ip-port": "2001:db8::10000"})
+    with raises(KresdManagerException):
+        TestSchema({"ip-port": "123.4.5.6@"})
+    with raises(KresdManagerException):
+        TestSchema({"ip-port": "123.4.5.6@-1"})
+    with raises(KresdManagerException):
+        TestSchema({"ip-port": "123.4.5.6@65536"})
+
+
 def test_ipaddress():
     class TestSchema(SchemaNode):
         ip: IPAddress