Add schema for network parameters (#388)

* Add NETWORK and NETWORK_UPDATE schemas.
* Refactor config.py
* Validate keys.
* Refactor update_network() signature.
Remove NETWORK_UPDATE schema.

* Add hex number validator/convertor.
* Add tests.
* Update coverage.

tests/test_config.py

@@ -3,6 +3,7 @@
 import pytest
 import voluptuous as vol
 import zigpy.config
+import zigpy.config.validators
 
 
 @pytest.mark.parametrize(
@@ -28,9 +29,9 @@
 )
 def test_config_validation_bool(value, result):
     """Test boolean config validation."""
-    assert zigpy.config.cv_boolean(value) is result
+    assert zigpy.config.validators.cv_boolean(value) is result
 
-    schema = vol.Schema({vol.Required("value"): zigpy.config.cv_boolean})
+    schema = vol.Schema({vol.Required("value"): zigpy.config.validators.cv_boolean})
     validated = schema({"value": value})
     assert validated["value"] is result
 
@@ -39,4 +40,88 @@ def test_config_validation_bool(value, result):
 def test_config_validation_bool_invalid(value):
     """Test boolean config validation."""
     with pytest.raises(vol.Invalid):
-        zigpy.config.cv_boolean(value)
+        zigpy.config.validators.cv_boolean(value)
+
+
+def test_config_validation_key_not_16_list():
+    """Validate key fails."""
+    with pytest.raises(vol.Invalid):
+        zigpy.config.validators.cv_key([0x00])
+
+    with pytest.raises(vol.Invalid):
+        zigpy.config.validators.cv_key([0x00 for i in range(15)])
+
+    with pytest.raises(vol.Invalid):
+        zigpy.config.validators.cv_key([0x00 for i in range(17)])
+
+    with pytest.raises(vol.Invalid):
+        zigpy.config.validators.cv_key(None)
+
+    zigpy.config.validators.cv_key([0x00 for i in range(16)])
+
+
+def test_config_validation_key_not_a_byte():
+    """Validate key fails."""
+
+    with pytest.raises(vol.Invalid):
+        zigpy.config.validators.cv_key([-1 for i in range(16)])
+
+    with pytest.raises(vol.Invalid):
+        zigpy.config.validators.cv_key([256 for i in range(16)])
+
+    zigpy.config.validators.cv_key([0xFF for i in range(16)])
+
+
+def test_config_validation_key_success():
+    """Validate key success."""
+
+    key = zigpy.config.validators.cv_key(zigpy.config.CONF_NWK_TC_LINK_KEY_DEFAULT)
+    assert key.serialize() == b"ZigBeeAlliance09"
+
+
+@pytest.mark.parametrize(
+    "value, result",
+    (
+        (0x1234, 0x1234),
+        ("0x1234", 0x1234),
+        (1234, 1234),
+        ("1234", 1234),
+        ("001234", 1234),
+        ("0e1234", vol.Invalid),
+        ("1234abcd", vol.Invalid),
+        ("0xabGG", vol.Invalid),
+        (None, vol.Invalid),
+    ),
+)
+def test_config_validation_hex_number(value, result):
+    """Test hex number config validation."""
+
+    if isinstance(result, int):
+        assert zigpy.config.validators.cv_hex(value) == result
+    else:
+        with pytest.raises(vol.Invalid):
+            zigpy.config.validators.cv_hex(value)
+
+
+@pytest.mark.parametrize(
+    "value, result",
+    (
+        (1, vol.Invalid),
+        (11, 11),
+        (0x11, 17),
+        ("26", 26),
+        (27, vol.Invalid),
+        ("27", vol.Invalid),
+    ),
+)
+def test_schema_network_channel(value, result):
+    """Test network schema for channel."""
+
+    config = {zigpy.config.CONF_NWK_CHANNEL: value}
+
+    if isinstance(result, int):
+        config = zigpy.config.SCHEMA_NETWORK(config)
+        assert config[zigpy.config.CONF_NWK_CHANNEL] == result
+    else:
+        with pytest.raises(vol.Invalid):
+            zigpy.config.SCHEMA_NETWORK(config)

zigpy/application.py

@@ -78,8 +78,8 @@ async def shutdown(self):
     async def startup(self, auto_form=False):
         """Perform a complete application startup"""
 
-    async def form_network(self, channel=15, pan_id=None, extended_pan_id=None):
-        """Form a new network"""
+    async def form_network(self):
+        """Form a new network based on network configuration from config."""
         raise NotImplementedError
 
     def add_device(self, ieee, nwk):
@@ -92,19 +92,24 @@ def add_device(self, ieee, nwk):
 
     async def update_network(
         self,
+        *,
         channel: Optional[t.uint8_t] = None,
-        channels: Optional[t.uint32_t] = None,
-        pan_id: Optional[t.PanId] = None,
+        channels: Optional[t.Channels] = None,
         extended_pan_id: Optional[t.ExtendedPanId] = None,
         network_key: Optional[t.KeyData] = None,
+        pan_id: Optional[t.PanId] = None,
+        tc_link_key: Optional[t.KeyData] = None,
+        update_id: int = 0,
     ):
         """Update network parameters.
 
         :param channel: Radio channel
         :param channels: Channels mask
-        :param pan_id: Network pan id
         :param extended_pan_id: Extended pan id
         :param network_key: network key
+        :param pan_id: Network pan id
+        :param tc_link_key: Trust Center link key
+        :param update_id: nwk_update_id parameter
         """
         raise NotImplementedError
 
@@ -205,7 +210,7 @@ async def mrequest(
         data,
         *,
         hops=0,
-        non_member_radius=3
+        non_member_radius=3,
     ):
         """Submit and send data out as a multicast transmission.
 

zigpy/config/__init__.py

@@ -0,0 +1,76 @@
+"""Config schemas and validation."""
+
+import voluptuous as vol
+from zigpy.config.defaults import (
+    CONF_NWK_CHANNEL_DEFAULT,
+    CONF_NWK_CHANNELS_DEFAULT,
+    CONF_NWK_EXTENDED_PAN_ID_DEFAULT,
+    CONF_NWK_KEY_DEFAULT,
+    CONF_NWK_PAN_ID_DEFAULT,
+    CONF_NWK_TC_LINK_KEY_DEFAULT,
+    CONF_NWK_UPDATE_ID_DEFAULT,
+    CONF_OTA_IKEA_DEFAULT,
+    CONF_OTA_LEDVANCE_DEFAULT,
+    CONF_OTA_OTAU_DIR_DEFAULT,
+)
+from zigpy.config.validators import cv_boolean, cv_hex, cv_key
+import zigpy.types as t
+
+CONF_DATABASE = "database_path"
+CONF_DEVICE = "device"
+CONF_DEVICE_PATH = "path"
+CONF_NWK = "network"
+CONF_NWK_CHANNEL = "channel"
+CONF_NWK_CHANNELS = "channels"
+CONF_NWK_EXTENDED_PAN_ID = "extended_pan_id"
+CONF_NWK_PAN_ID = "pan_id"
+CONF_NWK_KEY = "key"
+CONF_NWK_TC_LINK_KEY = "tc_link_key"
+CONF_NWK_UPDATE_ID = "update_id"
+CONF_OTA = "ota"
+CONF_OTA_DIR = "otau_directory"
+CONF_OTA_IKEA = "ikea_provider"
+CONF_OTA_LEDVANCE = "ledvance_provider"
+
+SCHEMA_DEVICE = vol.Schema({vol.Required(CONF_DEVICE_PATH): str})
+SCHEMA_NETWORK = vol.Schema(
+    {
+        vol.Optional(CONF_NWK_CHANNEL, default=CONF_NWK_CHANNEL_DEFAULT): vol.All(
+            cv_hex, vol.Range(min=11, max=26)
+        ),
+        vol.Optional(CONF_NWK_CHANNELS, default=CONF_NWK_CHANNELS_DEFAULT): vol.All(
+            list, t.Channels.from_channel_list
+        ),
+        vol.Optional(
+            CONF_NWK_EXTENDED_PAN_ID, default=CONF_NWK_EXTENDED_PAN_ID_DEFAULT
+        ): t.ExtendedPanId,
+        vol.Optional(CONF_NWK_KEY, default=CONF_NWK_KEY_DEFAULT): cv_key,
+        vol.Optional(CONF_NWK_PAN_ID, default=CONF_NWK_PAN_ID_DEFAULT): vol.All(
+            cv_hex, t.PanId
+        ),
+        vol.Optional(
+            CONF_NWK_TC_LINK_KEY, default=CONF_NWK_TC_LINK_KEY_DEFAULT
+        ): cv_key,
+        vol.Optional(CONF_NWK_UPDATE_ID, default=CONF_NWK_UPDATE_ID_DEFAULT): vol.All(
+            cv_hex, vol.Range(min=0, max=255)
+        ),
+    }
+)
+SCHEMA_OTA = {
+    vol.Optional(CONF_OTA_DIR, default=CONF_OTA_OTAU_DIR_DEFAULT): vol.Any(None, str),
+    vol.Optional(CONF_OTA_IKEA, default=CONF_OTA_IKEA_DEFAULT): cv_boolean,
+    vol.Optional(CONF_OTA_LEDVANCE, default=CONF_OTA_LEDVANCE_DEFAULT): cv_boolean,
+}
+
+ZIGPY_SCHEMA = vol.Schema(
+    {
+        vol.Optional(CONF_DATABASE, default=None): vol.Any(None, str),
+        vol.Optional(CONF_NWK, default={}): SCHEMA_NETWORK,
+        vol.Optional(CONF_OTA, default={}): SCHEMA_OTA,
+    },
+    extra=vol.ALLOW_EXTRA,
+)
+
+CONFIG_SCHEMA = ZIGPY_SCHEMA.extend(
+    {vol.Required(CONF_DEVICE): SCHEMA_DEVICE}, extra=vol.ALLOW_EXTRA
+)

zigpy/config/defaults.py

@@ -0,0 +1,52 @@
+from zigpy import types as t
+
+CONF_NWK_CHANNEL_DEFAULT = 15
+CONF_NWK_CHANNELS_DEFAULT = [15, 20, 25]
+CONF_NWK_EXTENDED_PAN_ID_DEFAULT = t.ExtendedPanId(
+    [0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]
+)
+CONF_NWK_PAN_ID_DEFAULT = t.PanId(0x0000)
+CONF_NWK_KEY_DEFAULT = t.KeyData(
+    [
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+        0x00,
+    ]
+)
+CONF_NWK_TC_LINK_KEY_DEFAULT = t.KeyData(
+    [
+        0x5A,
+        0x69,
+        0x67,
+        0x42,
+        0x65,
+        0x65,
+        0x41,
+        0x6C,
+        0x6C,
+        0x69,
+        0x61,
+        0x6E,
+        0x63,
+        0x65,
+        0x30,
+        0x39,
+    ]
+)
+CONF_NWK_UPDATE_ID_DEFAULT = 0x00
+CONF_OTA_IKEA_DEFAULT = False
+CONF_OTA_LEDVANCE_DEFAULT = False
+CONF_OTA_OTAU_DIR_DEFAULT = None

zigpy/config/validators.py

@@ -0,0 +1,52 @@
+from typing import List, Union
+
+import voluptuous as vol
+import zigpy.types as t
+
+
+def cv_boolean(value: Union[bool, int, str]) -> bool:
+    """Validate and coerce a boolean value."""
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        value = value.lower().strip()
+        if value in ("1", "true", "yes", "on", "enable"):
+            return True
+        if value in ("0", "false", "no", "off", "disable"):
+            return False
+    elif isinstance(value, int):
+        return bool(value)
+    raise vol.Invalid(f"invalid boolean '{value}' value")
+
+
+def cv_hex(value: Union[int, str]) -> int:
+    """Convert string with possible hex number into int."""
+    if isinstance(value, int):
+        return value
+
+    if not isinstance(value, str):
+        raise vol.Invalid(f"{value} is not a valid hex number")
+
+    try:
+        if value.startswith("0x"):
+            value = int(value, base=16)
+        else:
+            value = int(value)
+    except ValueError:
+        raise vol.Invalid(f"Could not convert '{value}' to number")
+
+    return value
+
+
+def cv_key(key: List[int]) -> t.KeyData:
+    """Validate a key."""
+    if not isinstance(key, list):
+        raise vol.Invalid("key is expected to be a list of integers")
+
+    if len(key) != 16:
+        raise vol.Invalid("key list length must be 16")
+
+    if not any((0 <= e <= 255 for e in key)):
+        raise vol.Invalid("Key element myst be a within (0..255) range")
+
+    return t.KeyData(key)