From: Aleš Date: Mon, 24 Jan 2022 16:09:20 +0000 (+0100) Subject: datamodel: types: types based on @port refactored X-Git-Tag: v6.0.0a1~45^2~8 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=384fc0982f8c0577da5ef346768b10034efba8a8;p=thirdparty%2Fknot-resolver.git datamodel: types: types based on @port refactored - added type for interface name --- diff --git a/manager/knot_resolver_manager/datamodel/types.py b/manager/knot_resolver_manager/datamodel/types.py index 4ed775db8..049a7c01d 100644 --- a/manager/knot_resolver_manager/datamodel/types.py +++ b/manager/knot_resolver_manager/datamodel/types.py @@ -1,15 +1,13 @@ import ipaddress import logging import re -from enum import Enum, auto from pathlib import Path from typing import Any, Dict, Optional, Pattern, Type, Union from typing_extensions import Literal from knot_resolver_manager.exceptions import SchemaException -from knot_resolver_manager.utils import CustomValueType, SchemaNode -from knot_resolver_manager.utils.modelling import Serializable +from knot_resolver_manager.utils import CustomValueType logger = logging.getLogger(__name__) @@ -281,6 +279,13 @@ class PortNumber(_IntRangeBase): _min: int = 1 _max: int = 65_535 + @classmethod + def from_str(cls: Type["PortNumber"], port: str, object_path: str = "/") -> "PortNumber": + try: + return cls(int(port), object_path) + except ValueError as e: + raise SchemaException(f"Invalid port number {port}", object_path) from e + class Unit(CustomValueType): _re: Pattern[str] @@ -430,52 +435,51 @@ class DomainName(_PatternBase): ) +class InterfaceName(_PatternBase): + _re = re.compile(r"^[a-zA-Z0-9]+(?:[-_][a-zA-Z0-9]+)*$") + + class InterfacePort(_StrCustomBase): - if_name: str + if_name: InterfaceName port: PortNumber def __init__(self, source_value: Any, object_path: str = "/") -> None: super().__init__(source_value) if isinstance(source_value, str): - if "@" in source_value: - parts = source_value.split("@", 1) - try: - self.port = PortNumber(int(parts[1])) - except ValueError as e: - raise SchemaException("Failed to parse port.", object_path) from e - self.if_name = parts[0] + parts = source_value.split("@") + if len(parts) == 2: + self.port = PortNumber.from_str(parts[1], object_path) + self.if_name = InterfaceName(parts[0], object_path) else: - raise SchemaException("Missing port number '@'.", object_path) + raise SchemaException(f"Expected '@', got '{source_value}'.", object_path) self._value = source_value else: raise SchemaException( - f"Unexpected value for '@'. Expected string, got '{source_value}'" - f" with type '{type(source_value)}'", + f"Unexpected value for '@'. " + f"Expected string, got '{source_value}' with type '{type(source_value)}'", object_path, ) class InterfaceOptionalPort(_StrCustomBase): - if_name: str + if_name: InterfaceName port: Optional[PortNumber] = None def __init__(self, source_value: Any, object_path: str = "/") -> None: super().__init__(source_value) if isinstance(source_value, str): - if "@" in source_value: - parts = source_value.split("@", 1) - try: - self.port = PortNumber(int(parts[1])) - except ValueError as e: - raise SchemaException("Failed to parse port.", object_path) from e - self.if_name = parts[0] + parts = source_value.split("@") + if 0 < len(parts) < 3: + self.if_name = InterfaceName(parts[0], object_path) + if len(parts) == 2: + self.port = PortNumber.from_str(parts[1], object_path) else: - self.if_name = source_value + raise SchemaException(f"Expected '[@]', got '{parts}'.", object_path) self._value = source_value else: raise SchemaException( - f"Unexpected value for a '[@]'. Expected string, got '{source_value}'" - f" with type '{type(source_value)}'", + f"Unexpected value for '[@]'. " + f"Expected string, got '{source_value}' with type '{type(source_value)}'", object_path, ) @@ -487,25 +491,20 @@ class IPAddressPort(_StrCustomBase): def __init__(self, source_value: Any, object_path: str = "/") -> None: super().__init__(source_value) if isinstance(source_value, str): - if "@" in source_value: - parts = source_value.split("@", 1) - try: - self.port = PortNumber(int(parts[1])) - except ValueError as e: - raise SchemaException("Failed to parse port.", object_path) from e - + parts = source_value.split("@") + if len(parts) == 2: + self.port = PortNumber.from_str(parts[1], object_path) try: self.addr = ipaddress.ip_address(parts[0]) except ValueError as e: - raise SchemaException("Failed to parse IP address.", object_path) from e + raise SchemaException(f"Failed to parse IP address '{parts[0]}'.", object_path) from e else: - raise SchemaException("Missing port number '@'.", object_path) + raise SchemaException(f"Expected '@', got '{source_value}'.", object_path) self._value = source_value - else: raise SchemaException( - f"Unexpected value for a '@'. Expected string, got '{source_value}'" - f" with type '{type(source_value)}'", + f"Unexpected value for '@'. " + f"Expected string, got '{source_value}' with type '{type(source_value)}'", object_path, ) @@ -516,28 +515,21 @@ class IPAddressOptionalPort(_StrCustomBase): def __init__(self, source_value: Any, object_path: str = "/") -> None: if isinstance(source_value, str): - if "@" in source_value: - parts = source_value.split("@", 1) - try: - self.port: Optional[PortNumber] = PortNumber(int(parts[1])) - except ValueError as e: - raise SchemaException("Failed to parse port.", object_path) from e - + parts = source_value.split("@") + if 0 < len(parts) < 3: try: self.addr = ipaddress.ip_address(parts[0]) except ValueError as e: - raise SchemaException("Failed to parse IP address.", object_path) from e + raise SchemaException(f"Failed to parse IP address '{parts[0]}'.", object_path) from e + if len(parts) == 2: + self.port = PortNumber.from_str(parts[1], object_path) else: - try: - self.addr = ipaddress.ip_address(source_value) - except ValueError as e: - raise SchemaException("Failed to parse IP address.", object_path) from e + raise SchemaException(f"Expected '[@]', got '{parts}'.", object_path) self._value = source_value - else: raise SchemaException( - f"Unexpected value for a '[@]'. Expected string, got '{source_value}'" - f" with type '{type(source_value)}'", + f"Unexpected value for a '[@]'. " + f"Expected string, got '{source_value}' with type '{type(source_value)}'", object_path, ) diff --git a/manager/tests/unit/datamodel/test_datamodel_types.py b/manager/tests/unit/datamodel/test_datamodel_types.py index c694608ae..d22ea0cfa 100644 --- a/manager/tests/unit/datamodel/test_datamodel_types.py +++ b/manager/tests/unit/datamodel/test_datamodel_types.py @@ -5,6 +5,7 @@ from pytest import raises from knot_resolver_manager.datamodel.types import ( CheckedPath, DomainName, + InterfaceName, InterfaceOptionalPort, InterfacePort, IPAddress, @@ -80,6 +81,27 @@ def test_domain_name(): TestSchema({"name": "b@d.domain.com."}) +def test_interface_name(): + assert InterfaceName("lo") + assert InterfaceName("eth0") + assert InterfaceName("wlo1") + assert InterfaceName("web_ifgrp") + assert InterfaceName("e8-2") + + with raises(KresManagerException): + InterfaceName("_lo") + with raises(KresManagerException): + InterfaceName("-wlo1") + with raises(KresManagerException): + InterfaceName("lo_") + with raises(KresManagerException): + InterfaceName("wlo1-") + with raises(KresManagerException): + InterfaceName("e8--2") + with raises(KresManagerException): + InterfaceName("web__ifgrp") + + def test_interface_port(): class TestSchema(SchemaNode): interface: InterfacePort