class InterfacePort(StrBase):
- addr: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
- if_name: InterfaceName
+ addr: Union[None, ipaddress.IPv4Address, ipaddress.IPv6Address] = None
+ if_name: Optional[InterfaceName] = None
port: PortNumber
def __init__(self, source_value: Any, object_path: str = "/") -> None:
from pytest import raises
from knot_resolver_manager.datamodel import KresConfig
-from knot_resolver_manager.datamodel.types import IPv6Network96, TimeUnit
from knot_resolver_manager.exceptions import SchemaException
from knot_resolver_manager.utils.modelling import SchemaNode
from tests.unit.utils import test_instance_of_kres_config
-def test_dns64_true():
- config = KresConfig({"server": {"id": "test"}, "dns64": True})
-
- assert config.dns64
- assert config.dns64.prefix == IPv6Network96("64:ff9b::/96")
-
-
-def test_dns64_default_false():
+def test_config_defaults():
config = test_instance_of_kres_config()
+ # DNS64 default
assert config.dns64 == False
def test_dnssec_default_true():
config = test_instance_of_kres_config()
+ # DNSSEC defaults
assert config.dnssec.trust_anchor_sentinel == True
assert config.dnssec.trust_anchor_signal_query == True
assert config.dnssec.time_skew_detection == True
- assert int(config.dnssec.keep_removed) == 0
assert config.dnssec.refresh_time == None
- assert config.dnssec.hold_down_time == TimeUnit("30d")
-
assert config.dnssec.trust_anchors == None
assert config.dnssec.negative_trust_anchors == None
assert config.dnssec.trust_anchors_files == None
+ assert int(config.dnssec.keep_removed) == 0
+ assert str(config.dnssec.hold_down_time) == "30d"
+
+
+def test_dns64_prefix_default():
+ assert str(KresConfig({"server": {"id": "test"}, "dns64": True}).dns64.prefix) == "64:ff9b::/96"
def test_json_schema():
+++ /dev/null
-import ipaddress
-
-from pytest import raises
-
-from knot_resolver_manager.datamodel.types import (
- CheckedPath,
- DomainName,
- InterfaceName,
- InterfaceOptionalPort,
- InterfacePort,
- IPAddress,
- IPAddressOptionalPort,
- IPAddressPort,
- IPNetwork,
- IPv4Address,
- IPv6Address,
- IPv6Network96,
- PortNumber,
- SizeUnit,
- TimeUnit,
-)
-from knot_resolver_manager.exceptions import KresManagerException
-from knot_resolver_manager.utils import SchemaNode
-
-
-def test_port_number():
- assert PortNumber(1)
- assert PortNumber(65_535)
- assert PortNumber(5353)
- assert PortNumber(5000)
-
- with raises(KresManagerException):
- PortNumber(0)
- with raises(KresManagerException):
- PortNumber(65_636)
- with raises(KresManagerException):
- PortNumber(-1)
-
-
-def test_size_unit():
- assert SizeUnit("5368709120B") == SizeUnit("5242880K") == SizeUnit("5120M") == SizeUnit("5G")
-
- with raises(KresManagerException):
- SizeUnit("-5368709120B")
- with raises(KresManagerException):
- SizeUnit(-5368709120)
- with raises(KresManagerException):
- SizeUnit("5120MM")
-
-
-def test_time_unit():
- assert TimeUnit("1d") == TimeUnit("24h") == TimeUnit("1440m") == TimeUnit("86400s")
-
- with raises(KresManagerException):
- TimeUnit("-1")
- with raises(KresManagerException):
- TimeUnit(-24)
- with raises(KresManagerException):
- TimeUnit("1440mm")
-
- assert TimeUnit("10ms").millis() == 10
-
-
-def test_parsing_units():
- class TestSchema(SchemaNode):
- size: SizeUnit
- time: TimeUnit
-
- o = TestSchema({"size": "3K", "time": "10m"})
- assert o.size == SizeUnit("3072B")
- assert o.time == TimeUnit("10m")
- assert o.size.bytes() == 3072
- assert o.time.seconds() == 10 * 60
-
-
-def test_checked_path():
- class TestSchema(SchemaNode):
- p: CheckedPath
-
- 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(KresManagerException):
- 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():
- o = InterfacePort("lo@5353")
- assert str(o) == "lo@5353"
- assert o == InterfacePort("lo@5353")
- assert str(o.if_name) == "lo"
- assert o.port == PortNumber(5353)
-
- o = InterfacePort("2001:db8::1000@5001")
- assert str(o) == "2001:db8::1000@5001"
- assert o == InterfacePort("2001:db8::1000@5001")
- assert str(o.addr) == "2001:db8::1000"
- assert o.port == PortNumber(5001)
-
- with raises(KresManagerException):
- InterfacePort("lo")
- with raises(KresManagerException):
- InterfacePort("53")
-
-
-def test_interface_optional_port():
- o = InterfaceOptionalPort("lo")
- assert str(o) == "lo"
- assert o == InterfaceOptionalPort("lo")
- assert str(o.if_name) == "lo"
- assert o.port == None
-
- o = InterfaceOptionalPort("123.4.5.6")
- assert str(o) == "123.4.5.6"
- assert o == InterfaceOptionalPort("123.4.5.6")
- assert str(o.addr) == "123.4.5.6"
- assert o.port == None
-
- o = InterfaceOptionalPort("lo@5353")
- assert str(o) == "lo@5353"
- assert o == InterfaceOptionalPort("lo@5353")
- assert str(o.if_name) == "lo"
- assert o.port == PortNumber(5353)
-
- o = InterfaceOptionalPort("2001:db8::1000@5001")
- assert str(o) == "2001:db8::1000@5001"
- assert o == InterfaceOptionalPort("2001:db8::1000@5001")
- assert str(o.addr) == "2001:db8::1000"
- assert o.port == PortNumber(5001)
-
- with raises(KresManagerException):
- InterfaceOptionalPort("lo@")
- with raises(KresManagerException):
- InterfaceOptionalPort("@53")
-
-
-def test_ip_address_port():
- class TestSchema(SchemaNode):
- ip_port: IPAddressPort
-
- 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@53"})
- assert str(o.ip_port) == "2001:db8::1000@53"
- assert o.ip_port == IPAddressPort("2001:db8::1000@53")
-
- with raises(KresManagerException):
- TestSchema({"ip-port": "123.4.5.6"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "2001:db8::1000"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "123.4.5.6.7@5000"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "2001:db8::10000@5001"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "123.4.5.6@"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "123.4.5.6@-1"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "123.4.5.6@65536"})
-
-
-def test_ip_address_optional_port():
- class TestSchema(SchemaNode):
- ip_port: IPAddressOptionalPort
-
- o = TestSchema({"ip-port": "123.4.5.6"})
- assert str(o.ip_port) == "123.4.5.6"
- assert o.ip_port == IPAddressOptionalPort("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 == IPAddressOptionalPort("123.4.5.6@5353")
-
- o = TestSchema({"ip-port": "2001:db8::1000"})
- assert str(o.ip_port) == "2001:db8::1000"
- assert o.ip_port == IPAddressOptionalPort("2001:db8::1000")
-
- o = TestSchema({"ip-port": "2001:db8::1000@53"})
- assert str(o.ip_port) == "2001:db8::1000@53"
- assert o.ip_port == IPAddressOptionalPort("2001:db8::1000@53")
-
- with raises(KresManagerException):
- TestSchema({"ip-port": "123.4.5.6.7"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "2001:db8::10000"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "123.4.5.6@"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "123.4.5.6@-1"})
- with raises(KresManagerException):
- TestSchema({"ip-port": "123.4.5.6@65536"})
-
-
-def test_ip_address():
- class TestSchema(SchemaNode):
- ip: IPAddress
-
- o = TestSchema({"ip": "123.4.5.6"})
- assert str(o.ip) == "123.4.5.6"
- assert o.ip == IPv4Address("123.4.5.6")
-
- o = TestSchema({"ip": "2001:db8::1000"})
- assert str(o.ip) == "2001:db8::1000"
- assert o.ip == IPv6Address("2001:db8::1000")
-
- with raises(KresManagerException):
- TestSchema({"ip": "123456"})
-
-
-def test_network():
- o = IPNetwork("10.11.12.0/24")
- assert o.to_std().prefixlen == 24
- assert o.to_std() == ipaddress.IPv4Network("10.11.12.0/24")
-
- with raises(KresManagerException):
- # because only the prefix can have non-zero bits
- IPNetwork("10.11.12.13/8")
-
-
-def test_ipv6_96_network():
- _ = IPv6Network96("fe80::/96")
-
- with raises(KresManagerException):
- IPv6Network96("fe80::/95")
-
- with raises(KresManagerException):
- IPv6Network96("10.11.12.3/96")
from knot_resolver_manager.datamodel.lua_schema import LuaSchema
from knot_resolver_manager.exceptions import KresManagerException
-tree = {"script-only": True, "script": "-- lua script"}
-strict = LuaSchema(tree)
-
-def test_validating():
- assert strict.script_only == True
- assert strict.script == "-- lua script"
-
-
-def test_exception_raises():
+def test_invalid():
with raises(KresManagerException):
LuaSchema({"script": "-- lua script", "script-file": "path/to/file"})
+from typing import Any, Dict, Optional
+
+import pytest
from pytest import raises
from knot_resolver_manager.datamodel.network_schema import ListenSchema, NetworkSchema
assert o.listen[1].freebind == True
-def test_listen_kind_port_defaults():
- assert ListenSchema({"unix-socket": "/tmp/kresd-socket"}).port == None
- assert ListenSchema({"interface": "::1"}).port == PortNumber(53)
- assert ListenSchema({"interface": "::1", "kind": "dot"}).port == PortNumber(853)
- assert ListenSchema({"interface": "::1", "kind": "doh-legacy"}).port == PortNumber(443)
- assert ListenSchema({"interface": "::1", "kind": "doh2"}).port == PortNumber(443)
-
-
-def test_listen_unix_socket_valid():
- assert ListenSchema({"unix-socket": "/tmp/kresd-socket"})
- assert ListenSchema({"unix-socket": ["/tmp/kresd-socket", "/tmp/kresd-socket2"]})
-
-
-def test_listen_unix_socket_invalid():
- with raises(KresManagerException):
- ListenSchema({"ip-address": "::1", "unix-socket": "/tmp/kresd-socket"})
- with raises(KresManagerException):
- ListenSchema({"unit-socket": "/tmp/kresd-socket", "port": "53"})
-
-
-def test_listen_ip_address_valid():
- assert ListenSchema({"interface": "::1"})
- assert ListenSchema({"interface": "::1@5353"})
- assert ListenSchema({"interface": "::1", "port": 5353})
- assert ListenSchema({"interface": ["127.0.0.1", "::1"]})
- assert ListenSchema({"interface": ["127.0.0.1@5353", "::1@5353"]})
- assert ListenSchema({"interface": ["127.0.0.1", "::1"], "port": 5353})
-
-
-def test_listen_ip_address_invalid():
- with raises(KresManagerException):
- ListenSchema({"ip-address": "::1@5353", "port": 5353})
- with raises(KresManagerException):
- ListenSchema({"ip-address": ["127.0.0.1", "::1@5353"]})
- with raises(KresManagerException):
- ListenSchema({"ip-address": ["127.0.0.1@5353", "::1@5353"], "port": 5353})
-
-
-def test_listen_interface_valid():
- assert ListenSchema({"interface": "lo"})
- assert ListenSchema({"interface": "lo@5353"})
- assert ListenSchema({"interface": "lo", "port": 5353})
- assert ListenSchema({"interface": ["lo", "eth0"]})
- assert ListenSchema({"interface": ["lo@5353", "eth0@5353"]})
- assert ListenSchema({"interface": ["lo", "eth0"], "port": 5353})
-
-
-def test_listen_interface_invalid():
- with raises(KresManagerException):
- ListenSchema({"interface": "lo@5353", "port": 5353})
- with raises(KresManagerException):
- ListenSchema({"interface": ["lo", "eth0@5353"]})
- with raises(KresManagerException):
- ListenSchema({"interface": ["lo@5353", "eth0@5353"], "port": 5353})
-
-
-def test_listen_invalid():
- with raises(KresManagerException):
- ListenSchema({"ip-address": "::1", "port": 0})
- with raises(KresManagerException):
- ListenSchema({"ip-address": "::1", "port": 65_536})
+@pytest.mark.parametrize(
+ "listen,port",
+ [
+ ({"unix-socket": "/tmp/kresd-socket"}, None),
+ ({"interface": "::1"}, 53),
+ ({"interface": "::1", "kind": "dot"}, 853),
+ ({"interface": "::1", "kind": "doh-legacy"}, 443),
+ ({"interface": "::1", "kind": "doh2"}, 443),
+ ],
+)
+def test_listen_port_defaults(listen: Dict[str, Any], port: Optional[int]):
+ assert ListenSchema(listen).port == (PortNumber(port) if port else None)
+
+
+@pytest.mark.parametrize(
+ "listen",
+ [
+ {"unix-socket": "/tmp/kresd-socket"},
+ {"unix-socket": ["/tmp/kresd-socket", "/tmp/kresd-socket2"]},
+ {"interface": "::1"},
+ {"interface": "::1@5353"},
+ {"interface": "::1", "port": 5353},
+ {"interface": ["127.0.0.1", "::1"]},
+ {"interface": ["127.0.0.1@5353", "::1@5353"]},
+ {"interface": ["127.0.0.1", "::1"], "port": 5353},
+ {"interface": "lo"},
+ {"interface": "lo@5353"},
+ {"interface": "lo", "port": 5353},
+ {"interface": ["lo", "eth0"]},
+ {"interface": ["lo@5353", "eth0@5353"]},
+ {"interface": ["lo", "eth0"], "port": 5353},
+ ],
+)
+def test_listen_valid(listen: Dict[str, Any]):
+ assert ListenSchema(listen)
+
+
+@pytest.mark.parametrize(
+ "listen",
+ [
+ {"unit-socket": "/tmp/kresd-socket", "port": "53"},
+ {"interface": "::1", "unit-socket": "/tmp/kresd-socket"},
+ {"interface": "::1@5353", "port": 5353},
+ {"interface": ["127.0.0.1", "::1@5353"]},
+ {"interface": ["127.0.0.1@5353", "::1@5353"], "port": 5353},
+ {"interface": "lo@5353", "port": 5353},
+ {"interface": ["lo", "eth0@5353"]},
+ {"interface": ["lo@5353", "eth0@5353"], "port": 5353},
+ ],
+)
+def test_listen_invalid(listen: Dict[str, Any]):
with raises(KresManagerException):
- ListenSchema({"ip-address": "::1", "interface": "lo"})
+ ListenSchema(listen)
from knot_resolver_manager.datamodel.options_schema import OptionsSchema
-from knot_resolver_manager.datamodel.types import IntNonNegative, TimeUnit
-def test_prediction_true():
- y = OptionsSchema({"prediction": True})
-
- assert y.prediction.window == TimeUnit("15m")
- assert int(y.prediction.period) == 24
+def test_prediction_true_defaults():
+ o = OptionsSchema({"prediction": True})
+ assert str(o.prediction.window) == "15m"
+ assert int(o.prediction.period) == 24
+from typing import Any, Dict
+
+import pytest
from pytest import raises
from knot_resolver_manager.datamodel.policy_schema import PolicySchema
from knot_resolver_manager.exceptions import KresManagerException
-def test_simple_actions():
- assert PolicySchema({"action": "pass"})
- assert PolicySchema({"action": "deny"})
- assert PolicySchema({"action": "drop"})
- assert PolicySchema({"action": "refuse"})
- assert PolicySchema({"action": "tc"})
- assert PolicySchema({"action": "debug-always"})
- assert PolicySchema({"action": "debug-cache-miss"})
- assert PolicySchema({"action": "qtrace"})
- assert PolicySchema({"action": "reqtrace"})
-
+@pytest.mark.parametrize("val", [{"action": "invalid-action"}])
+def test_simple_actions_invalid(val: Dict[str, Any]):
with raises(KresManagerException):
PolicySchema({"action": "invalid-action"})
-def test_deny_message():
- assert PolicySchema({"action": "deny", "message": "this is deny message"})
-
- with raises(KresManagerException):
- PolicySchema({"action": "pass", "message": "this is deny message"})
-
-
-def test_reroute():
- assert PolicySchema({"action": "reroute", "reroute": [{"source": "192.0.2.0/24", "destination": "127.0.0.0"}]})
-
- with raises(KresManagerException):
- PolicySchema({"action": "reroute"})
- with raises(KresManagerException):
- PolicySchema({"action": "pass", "reroute": [{"source": "192.0.2.0/24", "destination": "127.0.0.0"}]})
-
-
-def test_answer():
- assert PolicySchema({"action": "answer", "answer": {"rtype": "AAAA", "rdata": "::1"}})
-
- with raises(KresManagerException):
- PolicySchema({"action": "answer"})
- with raises(KresManagerException):
- PolicySchema({"action": "pass", "answer": {"rtype": "AAAA", "rdata": "::1"}})
-
-
-def test_mirror():
- assert PolicySchema({"action": "mirror", "mirror": ["127.0.0.1@5353"]})
-
- with raises(KresManagerException):
- PolicySchema({"action": "mirror"})
- with raises(KresManagerException):
- PolicySchema({"action": "pass", "mirror": ["127.0.0.1@5353"]})
+@pytest.mark.parametrize(
+ "val",
+ [
+ "pass",
+ "drop",
+ "refuse",
+ "tc",
+ "debug-always",
+ "debug-cache-miss",
+ "qtrace",
+ "reqtrace",
+ ],
+)
+def test_message_invalid(val: str):
+ with raises(KresManagerException):
+ PolicySchema({"action": f"{val}", "message": "this is deny message"})
+
+
+@pytest.mark.parametrize(
+ "val",
+ [
+ {"action": "reroute"},
+ {"action": "answer"},
+ {"action": "mirror"},
+ {"action": "pass", "reroute": [{"source": "192.0.2.0/24", "destination": "127.0.0.0"}]},
+ {"action": "pass", "answer": {"rtype": "AAAA", "rdata": "::1"}},
+ {"action": "pass", "mirror": ["127.0.0.1@5353"]},
+ ],
+)
+def test_invalid(val: Dict[str, Any]):
+ with raises(KresManagerException):
+ PolicySchema(val)
+import pytest
from pytest import raises
from knot_resolver_manager.datamodel.rpz_schema import RPZSchema
from knot_resolver_manager.exceptions import KresManagerException
-def test_message():
-
- assert RPZSchema({"action": "deny", "file": "blocklist.rpz", "message": "this is deny message"})
-
+@pytest.mark.parametrize(
+ "val",
+ [
+ "pass",
+ "drop",
+ "refuse",
+ "tc",
+ "debug-always",
+ "debug-cache-miss",
+ "qtrace",
+ "reqtrace",
+ ],
+)
+def test_message_invalid(val: str):
with raises(KresManagerException):
- RPZSchema({"action": "pass", "file": "whitelist.rpz", "message": "this is deny message"})
+ RPZSchema({"action": f"{val}", "file": "whitelist.rpz", "message": "this is deny message"})
-from pytest import raises
+from typing import Any, Dict, Optional
+
+import pytest
from knot_resolver_manager.datamodel.server_schema import ManagementSchema, ServerSchema
from knot_resolver_manager.exceptions import KresManagerException
-def test_watchdog():
- assert ServerSchema({"watchdog": {"qname": "nic.cz.", "qtype": "A"}, "id": "test"})
-
- with raises(KresManagerException):
+def test_watchdog_backend_invalid():
+ with pytest.raises(KresManagerException):
ServerSchema({"backend": "supervisord", "watchdog": {"qname": "nic.cz.", "qtype": "A"}})
-def test_management():
- assert ManagementSchema({"interface": "::1@53"})
- assert ManagementSchema({"unix-socket": "/path/socket"})
+@pytest.mark.parametrize("val", [{"interface": "::1@53"}, {"unix-socket": "/path/socket"}])
+def test_management_valid(val: Dict[str, Any]):
+ o = ManagementSchema(val)
+ if o.interface:
+ assert str(o.interface) == val["interface"]
+ if o.unix_socket:
+ assert str(o.unix_socket) == val["unix-socket"]
+
- with raises(KresManagerException):
- ManagementSchema()
- with raises(KresManagerException):
- ManagementSchema({"ip-address": "::1@53", "unix-socket": "/path/socket"})
+@pytest.mark.parametrize("val", [None, {"interface": "::1@53", "unix-socket": "/path/socket"}])
+def test_management_invalid(val: Optional[Dict[str, Any]]):
+ with pytest.raises(KresManagerException):
+ ManagementSchema(val)
--- /dev/null
+import ipaddress
+from typing import Any
+
+import pytest
+from pytest import raises
+
+from knot_resolver_manager.datamodel.types import (
+ CheckedPath,
+ DomainName,
+ InterfaceName,
+ InterfaceOptionalPort,
+ InterfacePort,
+ IPAddressOptionalPort,
+ IPAddressPort,
+ IPNetwork,
+ IPv4Address,
+ IPv6Address,
+ IPv6Network96,
+ PortNumber,
+ SizeUnit,
+ TimeUnit,
+)
+from knot_resolver_manager.exceptions import KresManagerException
+from knot_resolver_manager.utils import SchemaNode
+
+
+@pytest.mark.parametrize("val", [1, 65_535, 5353, 5000])
+def test_port_number_valid(val: int):
+ assert int(PortNumber(val)) == val
+
+
+@pytest.mark.parametrize("val", [0, 65_636, -1, "53"])
+def test_port_number_invalid(val: Any):
+ with raises(KresManagerException):
+ PortNumber(val)
+
+
+@pytest.mark.parametrize("val", ["5368709120B", "5242880K", "5120M", "5G"])
+def test_size_unit_valid(val: str):
+ o = SizeUnit(val)
+ assert int(o) == 5368709120
+ assert str(o) == val
+ assert o.bytes() == 5368709120
+
+
+@pytest.mark.parametrize("val", ["-5B", 5, -5242880, "45745mB"])
+def test_size_unit_invalid(val: Any):
+ with raises(KresManagerException):
+ SizeUnit(val)
+
+
+@pytest.mark.parametrize("val", ["1d", "24h", "1440m", "86400s", "86400000ms"])
+def test_time_unit_valid(val: str):
+ o = TimeUnit(val)
+ assert int(o) == 86400000
+ assert str(o) == val
+ assert o.seconds() == 86400
+ assert o.millis() == 86400000
+
+
+@pytest.mark.parametrize("val", ["-1", "-24h", "1440mm", 6575, -1440])
+def test_time_unit_invalid(val: Any):
+ with raises(KresManagerException):
+ TimeUnit("-1")
+
+
+def test_parsing_units():
+ class TestSchema(SchemaNode):
+ size: SizeUnit
+ time: TimeUnit
+
+ o = TestSchema({"size": "3K", "time": "10m"})
+ assert o.size == SizeUnit("3072B")
+ assert o.time == TimeUnit("10m")
+ assert o.size.bytes() == 3072
+ assert o.time.seconds() == 10 * 60
+
+
+def test_checked_path():
+ class TestSchema(SchemaNode):
+ p: CheckedPath
+
+ 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(KresManagerException):
+ TestSchema({"name": "b@d.domain.com."})
+
+
+@pytest.mark.parametrize("val", ["lo", "eth0", "wlo1", "web_ifgrp", "e8-2"])
+def test_interface_name_valid(val: str):
+ assert str(InterfaceName(val)) == val
+
+
+@pytest.mark.parametrize("val", ["_lo", "-wlo1", "lo_", "wlo1-", "e8--2", "web__ifgrp"])
+def test_interface_name_invalid(val: Any):
+ with raises(KresManagerException):
+ InterfaceName(val)
+
+
+@pytest.mark.parametrize("val", ["lo@5353", "2001:db8::1000@5001"])
+def test_interface_port_valid(val: str):
+ o = InterfacePort(val)
+ assert str(o) == val
+ assert o == InterfacePort(val)
+ assert str(o.if_name if o.if_name else o.addr) == val.split("@", 1)[0]
+ assert o.port == PortNumber(int(val.split("@", 1)[1]))
+
+
+@pytest.mark.parametrize("val", ["lo", "2001:db8::1000", "53"])
+def test_interface_port_invalid(val: Any):
+ with raises(KresManagerException):
+ InterfacePort(val)
+
+
+@pytest.mark.parametrize("val", ["lo", "123.4.5.6", "lo@5353", "2001:db8::1000@5001"])
+def test_interface_optional_port_valid(val: str):
+ o = InterfaceOptionalPort(val)
+ assert str(o) == val
+ assert o == InterfaceOptionalPort(val)
+ assert str(o.if_name if o.if_name else o.addr) == (val.split("@", 1)[0] if "@" in val else val)
+ assert o.port == (PortNumber(int(val.split("@", 1)[1])) if "@" in val else None)
+
+
+@pytest.mark.parametrize("val", ["lo@", "@53"])
+def test_interface_optional_port_invalid(val: Any):
+ with raises(KresManagerException):
+ InterfaceOptionalPort(val)
+
+
+@pytest.mark.parametrize("val", ["123.4.5.6@5353", "2001:db8::1000@53"])
+def test_ip_address_port_valid(val: str):
+ o = IPAddressPort(val)
+ assert str(o) == val
+ assert o == IPAddressPort(val)
+ assert str(o.addr) == val.split("@", 1)[0]
+ assert o.port == PortNumber(int(val.split("@", 1)[1]))
+
+
+@pytest.mark.parametrize(
+ "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(KresManagerException):
+ IPAddressPort(val)
+
+
+@pytest.mark.parametrize("val", ["123.4.5.6", "123.4.5.6@5353", "2001:db8::1000", "2001:db8::1000@53"])
+def test_ip_address_optional_port_valid(val: str):
+ o = IPAddressOptionalPort(val)
+ assert str(o) == val
+ assert o == IPAddressOptionalPort(val)
+ assert str(o.addr) == (val.split("@", 1)[0] if "@" in val else val)
+ assert o.port == (PortNumber(int(val.split("@", 1)[1])) if "@" in val else None)
+
+
+@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(KresManagerException):
+ IPAddressOptionalPort(val)
+
+
+@pytest.mark.parametrize("val", ["123.4.5.6", "192.168.0.1"])
+def test_ipv4_address_valid(val: str):
+ o = IPv4Address(val)
+ assert str(o) == val
+ assert o == IPv4Address(val)
+
+
+@pytest.mark.parametrize("val", ["123456", "2001:db8::1000"])
+def test_ipv4_address_invalid(val: Any):
+ with raises(KresManagerException):
+ IPv4Address(val)
+
+
+@pytest.mark.parametrize(
+ "val",
+ [
+ "2001:db8::1000",
+ ],
+)
+def test_ipv6_address_valid(val: str):
+ o = IPv6Address(val)
+ assert str(o) == val
+ assert o == IPv6Address(val)
+
+
+@pytest.mark.parametrize("val", ["123.4.5.6", "2001::db8::1000"])
+def test_ipv6_address_invalid(val: Any):
+ with raises(KresManagerException):
+ IPv6Address(val)
+
+
+@pytest.mark.parametrize("val", ["10.11.12.0/24", "64:ff9b::/96"])
+def test_ip_network_valid(val: str):
+ o = IPNetwork(val)
+ assert str(o) == val
+ assert o.to_std().prefixlen == int(val.split("/", 1)[1])
+ assert o.to_std() == ipaddress.ip_network(val)
+
+
+@pytest.mark.parametrize("val", ["10.11.12.13/8", "10.11.12.5/128"])
+def test_ip_network_invalid(val: str):
+ with raises(KresManagerException):
+ # because only the prefix can have non-zero bits
+ IPNetwork(val)
+
+
+@pytest.mark.parametrize("val", ["fe80::/96", "64:ff9b::/96"])
+def test_ipv6_96_network_valid(val: str):
+ assert str(IPv6Network96(val)) == val
+
+
+@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(KresManagerException):
+ IPv6Network96(val)
projects / thirdparty / knot-resolver.git / commitdiff
