Support for deconz discovery & cleanup (#974)

* Support for deconz discovery & cleanup

* Split discovery

* Fix lint

* Fix lint / import
This commit is contained in:
Pascal Vizeli 2019-03-28 14:11:18 +01:00 committed by GitHub
parent b52f90187b
commit b12175ab9a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 441 additions and 229 deletions

View File

@ -21,7 +21,7 @@ from ..const import (
ATTR_UUID, ATTR_VERSION, ATTR_WEBUI, BOOT_AUTO, BOOT_MANUAL,
PRIVILEGED_ALL, ROLE_ALL, ROLE_DEFAULT, STARTUP_ALL, STARTUP_APPLICATION,
STARTUP_SERVICES, STATE_STARTED, STATE_STOPPED)
from ..services.validate import DISCOVERY_SERVICES
from ..discovery.validate import valid_discovery_service
from ..validate import (
ALSA_DEVICE, DOCKER_PORTS, NETWORK_PORT, SHA256, UUID_MATCH)
@ -114,7 +114,7 @@ SCHEMA_ADDON_CONFIG = vol.Schema({
vol.Optional(ATTR_DOCKER_API, default=False): vol.Boolean(),
vol.Optional(ATTR_AUTH_API, default=False): vol.Boolean(),
vol.Optional(ATTR_SERVICES): [vol.Match(RE_SERVICE)],
vol.Optional(ATTR_DISCOVERY): [vol.In(DISCOVERY_SERVICES)],
vol.Optional(ATTR_DISCOVERY): [valid_discovery_service],
vol.Required(ATTR_OPTIONS): dict,
vol.Required(ATTR_SCHEMA): vol.Any(vol.Schema({
vol.Coerce(str): vol.Any(SCHEMA_ELEMENT, [

View File

@ -3,17 +3,24 @@ import voluptuous as vol
from .utils import api_process, api_validate
from ..const import (
ATTR_ADDON, ATTR_UUID, ATTR_CONFIG, ATTR_DISCOVERY, ATTR_SERVICE,
REQUEST_FROM)
ATTR_ADDON,
ATTR_UUID,
ATTR_CONFIG,
ATTR_DISCOVERY,
ATTR_SERVICE,
REQUEST_FROM,
)
from ..coresys import CoreSysAttributes
from ..exceptions import APIError, APIForbidden
from ..validate import SERVICE_ALL
from ..discovery.validate import valid_discovery_service
SCHEMA_DISCOVERY = vol.Schema({
vol.Required(ATTR_SERVICE): SERVICE_ALL,
vol.Optional(ATTR_CONFIG): vol.Maybe(dict),
})
SCHEMA_DISCOVERY = vol.Schema(
{
vol.Required(ATTR_SERVICE): valid_discovery_service,
vol.Optional(ATTR_CONFIG): vol.Maybe(dict),
}
)
class APIDiscovery(CoreSysAttributes):
@ -21,7 +28,7 @@ class APIDiscovery(CoreSysAttributes):
def _extract_message(self, request):
"""Extract discovery message from URL."""
message = self.sys_discovery.get(request.match_info.get('uuid'))
message = self.sys_discovery.get(request.match_info.get("uuid"))
if not message:
raise APIError("Discovery message not found")
return message
@ -38,12 +45,14 @@ class APIDiscovery(CoreSysAttributes):
discovery = []
for message in self.sys_discovery.list_messages:
discovery.append({
ATTR_ADDON: message.addon,
ATTR_SERVICE: message.service,
ATTR_UUID: message.uuid,
ATTR_CONFIG: message.config,
})
discovery.append(
{
ATTR_ADDON: message.addon,
ATTR_SERVICE: message.service,
ATTR_UUID: message.uuid,
ATTR_CONFIG: message.config,
}
)
return {ATTR_DISCOVERY: discovery}

View File

@ -159,7 +159,6 @@ ATTR_ADDON = "addon"
ATTR_AVAILABLE = "available"
ATTR_HOST = "host"
ATTR_USERNAME = "username"
ATTR_PROTOCOL = "protocol"
ATTR_DISCOVERY = "discovery"
ATTR_CONFIG = "config"
ATTR_SERVICES = "services"
@ -189,7 +188,6 @@ ATTR_AUTH_API = "auth_api"
ATTR_KERNEL_MODULES = "kernel_modules"
ATTR_SUPPORTED_ARCH = "supported_arch"
SERVICE_MQTT = "mqtt"
PROVIDE_SERVICE = "provide"
NEED_SERVICE = "need"
WANT_SERVICE = "want"

View File

@ -1,35 +1,50 @@
"""Handle discover message for Home Assistant."""
import logging
from __future__ import annotations
from contextlib import suppress
from uuid import uuid4
import logging
from typing import Any, Dict, List, Optional, TYPE_CHECKING
from uuid import uuid4, UUID
import attr
import voluptuous as vol
from voluptuous.humanize import humanize_error
from .const import FILE_HASSIO_DISCOVERY, ATTR_CONFIG, ATTR_DISCOVERY
from .coresys import CoreSysAttributes
from .exceptions import DiscoveryError, HomeAssistantAPIError
from .validate import SCHEMA_DISCOVERY_CONFIG
from .utils.json import JsonConfig
from .services.validate import DISCOVERY_SERVICES
from ..const import ATTR_CONFIG, ATTR_DISCOVERY, FILE_HASSIO_DISCOVERY
from ..coresys import CoreSys, CoreSysAttributes
from ..exceptions import DiscoveryError, HomeAssistantAPIError
from ..utils.json import JsonConfig
from .validate import SCHEMA_DISCOVERY_CONFIG, valid_discovery_config
if TYPE_CHECKING:
from ..addons.addon import Addon
_LOGGER = logging.getLogger(__name__)
CMD_NEW = 'post'
CMD_DEL = 'delete'
CMD_NEW = "post"
CMD_DEL = "delete"
@attr.s
class Message:
"""Represent a single Discovery message."""
addon: str = attr.ib()
service: str = attr.ib()
config: Dict[str, Any] = attr.ib(cmp=False)
uuid: UUID = attr.ib(factory=lambda: uuid4().hex, cmp=False)
class Discovery(CoreSysAttributes, JsonConfig):
"""Home Assistant Discovery handler."""
def __init__(self, coresys):
def __init__(self, coresys: CoreSys):
"""Initialize discovery handler."""
super().__init__(FILE_HASSIO_DISCOVERY, SCHEMA_DISCOVERY_CONFIG)
self.coresys = coresys
self.message_obj = {}
self.coresys: CoreSys = coresys
self.message_obj: Dict[str, Message] = {}
async def load(self):
async def load(self) -> None:
"""Load exists discovery message into storage."""
messages = {}
for message in self._data[ATTR_DISCOVERY]:
@ -39,9 +54,9 @@ class Discovery(CoreSysAttributes, JsonConfig):
_LOGGER.info("Load %d messages", len(messages))
self.message_obj = messages
def save(self):
def save(self) -> None:
"""Write discovery message into data file."""
messages = []
messages: List[Dict[str, Any]] = []
for message in self.list_messages:
messages.append(attr.asdict(message))
@ -49,22 +64,21 @@ class Discovery(CoreSysAttributes, JsonConfig):
self._data[ATTR_DISCOVERY].extend(messages)
self.save_data()
def get(self, uuid):
def get(self, uuid: str) -> Optional[Message]:
"""Return discovery message."""
return self.message_obj.get(uuid)
@property
def list_messages(self):
def list_messages(self) -> List[Message]:
"""Return list of available discovery messages."""
return list(self.message_obj.values())
def send(self, addon, service, config):
def send(self, addon: Addon, service: str, config: Dict[str, Any]) -> Message:
"""Send a discovery message to Home Assistant."""
try:
config = DISCOVERY_SERVICES[service](config)
config = valid_discovery_config(service, config)
except vol.Invalid as err:
_LOGGER.error(
"Invalid discovery %s config", humanize_error(config, err))
_LOGGER.error("Invalid discovery %s config", humanize_error(config, err))
raise DiscoveryError() from None
# Create message
@ -77,24 +91,26 @@ class Discovery(CoreSysAttributes, JsonConfig):
_LOGGER.info("Duplicate discovery message from %s", addon.slug)
return old_message
_LOGGER.info("Send discovery to Home Assistant %s from %s",
service, addon.slug)
_LOGGER.info("Send discovery to Home Assistant %s from %s", service, addon.slug)
self.message_obj[message.uuid] = message
self.save()
self.sys_create_task(self._push_discovery(message, CMD_NEW))
return message
def remove(self, message):
def remove(self, message: Message) -> None:
"""Remove a discovery message from Home Assistant."""
self.message_obj.pop(message.uuid, None)
self.save()
_LOGGER.info("Delete discovery to Home Assistant %s from %s",
message.service, message.addon)
_LOGGER.info(
"Delete discovery to Home Assistant %s from %s",
message.service,
message.addon,
)
self.sys_create_task(self._push_discovery(message, CMD_DEL))
async def _push_discovery(self, message, command):
async def _push_discovery(self, message: Message, command: str) -> None:
"""Send a discovery request."""
if not await self.sys_homeassistant.check_api_state():
_LOGGER.info("Discovery %s mesage ignore", message.uuid)
@ -105,18 +121,12 @@ class Discovery(CoreSysAttributes, JsonConfig):
with suppress(HomeAssistantAPIError):
async with self.sys_homeassistant.make_request(
command, f"api/hassio_push/discovery/{message.uuid}",
json=data, timeout=10):
command,
f"api/hassio_push/discovery/{message.uuid}",
json=data,
timeout=10,
):
_LOGGER.info("Discovery %s message send", message.uuid)
return
_LOGGER.warning("Discovery %s message fail", message.uuid)
@attr.s
class Message:
"""Represent a single Discovery message."""
addon = attr.ib()
service = attr.ib()
config = attr.ib(cmp=False)
uuid = attr.ib(factory=lambda: uuid4().hex, cmp=False)

View File

@ -0,0 +1,8 @@
"""Discovery static data."""
ATTR_HOST = "host"
ATTR_PASSWORD = "password"
ATTR_PORT = "port"
ATTR_PROTOCOL = "protocol"
ATTR_SSL = "ssl"
ATTR_USERNAME = "username"

View File

@ -0,0 +1 @@
"""Discovery service modules."""

View File

@ -0,0 +1,11 @@
"""Discovery service for MQTT."""
import voluptuous as vol
from hassio.validate import NETWORK_PORT
from ..const import ATTR_HOST, ATTR_PORT
# pylint: disable=no-value-for-parameter
SCHEMA = vol.Schema(
{vol.Required(ATTR_HOST): vol.Coerce(str), vol.Required(ATTR_PORT): NETWORK_PORT}
)

View File

@ -0,0 +1,27 @@
"""Discovery service for MQTT."""
import voluptuous as vol
from hassio.validate import NETWORK_PORT
from ..const import (
ATTR_HOST,
ATTR_PASSWORD,
ATTR_PORT,
ATTR_PROTOCOL,
ATTR_SSL,
ATTR_USERNAME,
)
# pylint: disable=no-value-for-parameter
SCHEMA = vol.Schema(
{
vol.Required(ATTR_HOST): vol.Coerce(str),
vol.Required(ATTR_PORT): NETWORK_PORT,
vol.Optional(ATTR_USERNAME): vol.Coerce(str),
vol.Optional(ATTR_PASSWORD): vol.Coerce(str),
vol.Optional(ATTR_SSL, default=False): vol.Boolean(),
vol.Optional(ATTR_PROTOCOL, default="3.1.1"): vol.All(
vol.Coerce(str), vol.In(["3.1", "3.1.1"])
),
}
)

View File

@ -0,0 +1,47 @@
"""Validate services schema."""
from pathlib import Path
from importlib import import_module
import voluptuous as vol
from ..const import ATTR_ADDON, ATTR_CONFIG, ATTR_DISCOVERY, ATTR_SERVICE, ATTR_UUID
from ..utils.validate import schema_or
from ..validate import UUID_MATCH
def valid_discovery_service(service):
"""Validate service name."""
service_file = Path(__file__).parent.joinpath(f"services/{service}.py")
if not service_file.exists():
raise vol.Invalid(f"Service {service} not found")
return service
def valid_discovery_config(service, config):
"""Validate service name."""
try:
service_mod = import_module(f".services.{service}", "hassio.discovery")
except ImportError:
raise vol.Invalid(f"Service {service} not found")
return service_mod.SCHEMA(config)
SCHEMA_DISCOVERY = vol.Schema(
[
vol.Schema(
{
vol.Required(ATTR_UUID): UUID_MATCH,
vol.Required(ATTR_ADDON): vol.Coerce(str),
vol.Required(ATTR_SERVICE): valid_discovery_service,
vol.Required(ATTR_CONFIG): vol.Maybe(dict),
},
extra=vol.REMOVE_EXTRA,
)
]
)
SCHEMA_DISCOVERY_CONFIG = vol.Schema(
{vol.Optional(ATTR_DISCOVERY, default=list): schema_or(SCHEMA_DISCOVERY)},
extra=vol.REMOVE_EXTRA,
)

View File

@ -1,38 +1,38 @@
"""Handle internal services discovery."""
from .mqtt import MQTTService
from typing import Dict, List
from ..coresys import CoreSys, CoreSysAttributes
from .const import SERVICE_MQTT
from .data import ServicesData
from ..const import SERVICE_MQTT
from ..coresys import CoreSysAttributes
from .interface import ServiceInterface
from .modules.mqtt import MQTTService
AVAILABLE_SERVICES = {
SERVICE_MQTT: MQTTService
}
AVAILABLE_SERVICES = {SERVICE_MQTT: MQTTService}
class ServiceManager(CoreSysAttributes):
"""Handle internal services discovery."""
def __init__(self, coresys):
def __init__(self, coresys: CoreSys):
"""Initialize Services handler."""
self.coresys = coresys
self.data = ServicesData()
self.services_obj = {}
self.coresys: CoreSys = coresys
self.data: ServicesData = ServicesData()
self.services_obj: Dict[str, ServiceInterface] = {}
@property
def list_services(self):
def list_services(self) -> List[ServiceInterface]:
"""Return a list of services."""
return list(self.services_obj.values())
def get(self, slug):
def get(self, slug: str) -> ServiceInterface:
"""Return service object from slug."""
return self.services_obj.get(slug)
async def load(self):
async def load(self) -> None:
"""Load available services."""
for slug, service in AVAILABLE_SERVICES.items():
self.services_obj[slug] = service(self.coresys)
def reset(self):
def reset(self) -> None:
"""Reset available data."""
self.data.reset_data()

11
hassio/services/const.py Normal file
View File

@ -0,0 +1,11 @@
"""Service API static data."""
ATTR_ADDON = "addon"
ATTR_HOST = "host"
ATTR_PASSWORD = "password"
ATTR_PORT = "port"
ATTR_PROTOCOL = "protocol"
ATTR_SSL = "ssl"
ATTR_USERNAME = "username"
SERVICE_MQTT = "mqtt"

View File

@ -1,8 +1,10 @@
"""Handle service data for persistent supervisor reboot."""
from typing import Any, Dict
from .validate import SCHEMA_SERVICES_CONFIG
from ..const import FILE_HASSIO_SERVICES, SERVICE_MQTT
from ..const import FILE_HASSIO_SERVICES
from ..utils.json import JsonConfig
from .const import SERVICE_MQTT
from .validate import SCHEMA_SERVICES_CONFIG
class ServicesData(JsonConfig):
@ -13,6 +15,6 @@ class ServicesData(JsonConfig):
super().__init__(FILE_HASSIO_SERVICES, SCHEMA_SERVICES_CONFIG)
@property
def mqtt(self):
def mqtt(self) -> Dict[str, Any]:
"""Return settings for MQTT service."""
return self._data[SERVICE_MQTT]

View File

@ -1,33 +1,37 @@
"""Interface for single service."""
from typing import Any, Dict, List, Optional
from ..coresys import CoreSysAttributes
import voluptuous as vol
from ..addons.addon import Addon
from ..const import PROVIDE_SERVICE
from ..coresys import CoreSys, CoreSysAttributes
class ServiceInterface(CoreSysAttributes):
"""Interface class for service integration."""
def __init__(self, coresys):
def __init__(self, coresys: CoreSys):
"""Initialize service interface."""
self.coresys = coresys
self.coresys: CoreSys = coresys
@property
def slug(self):
def slug(self) -> str:
"""Return slug of this service."""
return None
raise NotImplementedError()
@property
def _data(self):
def _data(self) -> Dict[str, Any]:
"""Return data of this service."""
return None
raise NotImplementedError()
@property
def schema(self):
def schema(self) -> vol.Schema:
"""Return data schema of this service."""
return None
raise NotImplementedError()
@property
def providers(self):
def providers(self) -> List[str]:
"""Return name of service providers addon."""
addons = []
for addon in self.sys_addons.list_installed:
@ -36,24 +40,24 @@ class ServiceInterface(CoreSysAttributes):
return addons
@property
def enabled(self):
def enabled(self) -> bool:
"""Return True if the service is in use."""
return bool(self._data)
def save(self):
def save(self) -> None:
"""Save changes."""
self.sys_services.data.save_data()
def get_service_data(self):
def get_service_data(self) -> Optional[Dict[str, Any]]:
"""Return the requested service data."""
if self.enabled:
return self._data
return None
def set_service_data(self, addon, data):
def set_service_data(self, addon: Addon, data: Dict[str, Any]) -> None:
"""Write the data into service object."""
raise NotImplementedError()
def del_service_data(self, addon):
def del_service_data(self, addon: Addon) -> None:
"""Remove the data from service object."""
raise NotImplementedError()

View File

@ -0,0 +1 @@
"""Services modules."""

View File

@ -0,0 +1,81 @@
"""Provide the MQTT Service."""
import logging
from typing import Any, Dict
from hassio.addons.addon import Addon
from hassio.exceptions import ServicesError
from hassio.validate import NETWORK_PORT
import voluptuous as vol
from ..const import (
ATTR_ADDON,
ATTR_HOST,
ATTR_PASSWORD,
ATTR_PORT,
ATTR_PROTOCOL,
ATTR_SSL,
ATTR_USERNAME,
SERVICE_MQTT,
)
from ..interface import ServiceInterface
_LOGGER = logging.getLogger(__name__)
# pylint: disable=no-value-for-parameter
SCHEMA_SERVICE_MQTT = vol.Schema(
{
vol.Required(ATTR_HOST): vol.Coerce(str),
vol.Required(ATTR_PORT): NETWORK_PORT,
vol.Optional(ATTR_USERNAME): vol.Coerce(str),
vol.Optional(ATTR_PASSWORD): vol.Coerce(str),
vol.Optional(ATTR_SSL, default=False): vol.Boolean(),
vol.Optional(ATTR_PROTOCOL, default="3.1.1"): vol.All(
vol.Coerce(str), vol.In(["3.1", "3.1.1"])
),
}
)
SCHEMA_CONFIG_MQTT = SCHEMA_SERVICE_MQTT.extend(
{vol.Required(ATTR_ADDON): vol.Coerce(str)}
)
class MQTTService(ServiceInterface):
"""Provide MQTT services."""
@property
def slug(self) -> str:
"""Return slug of this service."""
return SERVICE_MQTT
@property
def _data(self) -> Dict[str, Any]:
"""Return data of this service."""
return self.sys_services.data.mqtt
@property
def schema(self) -> vol.Schema:
"""Return data schema of this service."""
return SCHEMA_SERVICE_MQTT
def set_service_data(self, addon: Addon, data: Dict[str, Any]) -> None:
"""Write the data into service object."""
if self.enabled:
_LOGGER.error("It is already a MQTT in use from %s", self._data[ATTR_ADDON])
raise ServicesError()
self._data.update(data)
self._data[ATTR_ADDON] = addon.slug
_LOGGER.info("Set %s as service provider for mqtt", addon.slug)
self.save()
def del_service_data(self, addon: Addon) -> None:
"""Remove the data from service object."""
if not self.enabled:
_LOGGER.warning("Can't remove not exists services")
raise ServicesError()
self._data.clear()
self.save()

View File

@ -1,50 +0,0 @@
"""Provide the MQTT Service."""
import logging
from .interface import ServiceInterface
from .validate import SCHEMA_SERVICE_MQTT
from ..const import ATTR_ADDON, SERVICE_MQTT
from ..exceptions import ServicesError
_LOGGER = logging.getLogger(__name__)
class MQTTService(ServiceInterface):
"""Provide MQTT services."""
@property
def slug(self):
"""Return slug of this service."""
return SERVICE_MQTT
@property
def _data(self):
"""Return data of this service."""
return self.sys_services.data.mqtt
@property
def schema(self):
"""Return data schema of this service."""
return SCHEMA_SERVICE_MQTT
def set_service_data(self, addon, data):
"""Write the data into service object."""
if self.enabled:
_LOGGER.error(
"It is already a MQTT in use from %s", self._data[ATTR_ADDON])
raise ServicesError()
self._data.update(data)
self._data[ATTR_ADDON] = addon.slug
_LOGGER.info("Set %s as service provider for mqtt", addon.slug)
self.save()
def del_service_data(self, addon):
"""Remove the data from service object."""
if not self.enabled:
_LOGGER.warning("Can't remove not exists services")
raise ServicesError()
self._data.clear()
self.save()

View File

@ -1,35 +1,12 @@
"""Validate services schema."""
import voluptuous as vol
from ..const import (
SERVICE_MQTT, ATTR_HOST, ATTR_PORT, ATTR_PASSWORD, ATTR_USERNAME, ATTR_SSL,
ATTR_ADDON, ATTR_PROTOCOL)
from ..validate import NETWORK_PORT
from ..utils.validate import schema_or
from .const import SERVICE_MQTT
from .modules.mqtt import SCHEMA_CONFIG_MQTT
# pylint: disable=no-value-for-parameter
SCHEMA_SERVICE_MQTT = vol.Schema({
vol.Required(ATTR_HOST): vol.Coerce(str),
vol.Required(ATTR_PORT): NETWORK_PORT,
vol.Optional(ATTR_USERNAME): vol.Coerce(str),
vol.Optional(ATTR_PASSWORD): vol.Coerce(str),
vol.Optional(ATTR_SSL, default=False): vol.Boolean(),
vol.Optional(ATTR_PROTOCOL, default='3.1.1'):
vol.All(vol.Coerce(str), vol.In(['3.1', '3.1.1'])),
})
SCHEMA_CONFIG_MQTT = SCHEMA_SERVICE_MQTT.extend({
vol.Required(ATTR_ADDON): vol.Coerce(str),
})
SCHEMA_SERVICES_CONFIG = vol.Schema({
vol.Optional(SERVICE_MQTT, default=dict): schema_or(SCHEMA_CONFIG_MQTT),
}, extra=vol.REMOVE_EXTRA)
DISCOVERY_SERVICES = {
SERVICE_MQTT: SCHEMA_SERVICE_MQTT,
}
SCHEMA_SERVICES_CONFIG = vol.Schema(
{vol.Optional(SERVICE_MQTT, default=dict): schema_or(SCHEMA_CONFIG_MQTT)},
extra=vol.REMOVE_EXTRA,
)

View File

@ -5,14 +5,30 @@ import re
import voluptuous as vol
from .const import (
ATTR_IMAGE, ATTR_LAST_VERSION, ATTR_CHANNEL, ATTR_TIMEZONE, ATTR_HASSOS,
ATTR_ADDONS_CUSTOM_LIST, ATTR_PASSWORD, ATTR_HOMEASSISTANT, ATTR_HASSIO,
ATTR_BOOT, ATTR_LAST_BOOT, ATTR_SSL, ATTR_PORT, ATTR_WATCHDOG, ATTR_CONFIG,
ATTR_WAIT_BOOT, ATTR_UUID, ATTR_REFRESH_TOKEN, ATTR_HASSOS_CLI,
ATTR_ACCESS_TOKEN, ATTR_DISCOVERY, ATTR_ADDON, ATTR_SERVICE,
SERVICE_MQTT,
CHANNEL_STABLE, CHANNEL_BETA, CHANNEL_DEV)
from .utils.validate import schema_or, validate_timezone
ATTR_IMAGE,
ATTR_LAST_VERSION,
ATTR_CHANNEL,
ATTR_TIMEZONE,
ATTR_HASSOS,
ATTR_ADDONS_CUSTOM_LIST,
ATTR_PASSWORD,
ATTR_HOMEASSISTANT,
ATTR_HASSIO,
ATTR_BOOT,
ATTR_LAST_BOOT,
ATTR_SSL,
ATTR_PORT,
ATTR_WATCHDOG,
ATTR_WAIT_BOOT,
ATTR_UUID,
ATTR_REFRESH_TOKEN,
ATTR_HASSOS_CLI,
ATTR_ACCESS_TOKEN,
CHANNEL_STABLE,
CHANNEL_BETA,
CHANNEL_DEV,
)
from .utils.validate import validate_timezone
RE_REPOSITORY = re.compile(r"^(?P<url>[^#]+)(?:#(?P<branch>[\w\-]+))?$")
@ -24,7 +40,6 @@ ALSA_DEVICE = vol.Maybe(vol.Match(r"\d+,\d+"))
CHANNELS = vol.In([CHANNEL_STABLE, CHANNEL_BETA, CHANNEL_DEV])
UUID_MATCH = vol.Match(r"^[0-9a-f]{32}$")
SHA256 = vol.Match(r"^[0-9a-f]{64}$")
SERVICE_ALL = vol.In([SERVICE_MQTT])
def validate_repository(repository):
@ -35,7 +50,7 @@ def validate_repository(repository):
# Validate URL
# pylint: disable=no-value-for-parameter
vol.Url()(data.group('url'))
vol.Url()(data.group("url"))
return repository
@ -66,63 +81,61 @@ def convert_to_docker_ports(data):
raise vol.Invalid("Can't validate Docker host settings")
DOCKER_PORTS = vol.Schema({
vol.All(vol.Coerce(str), vol.Match(r"^\d+(?:/tcp|/udp)?$")):
convert_to_docker_ports,
})
DOCKER_PORTS = vol.Schema(
{
vol.All(
vol.Coerce(str), vol.Match(r"^\d+(?:/tcp|/udp)?$")
): convert_to_docker_ports
}
)
# pylint: disable=no-value-for-parameter
SCHEMA_HASS_CONFIG = vol.Schema({
vol.Optional(ATTR_UUID, default=lambda: uuid.uuid4().hex): UUID_MATCH,
vol.Optional(ATTR_ACCESS_TOKEN): SHA256,
vol.Optional(ATTR_BOOT, default=True): vol.Boolean(),
vol.Inclusive(ATTR_IMAGE, 'custom_hass'): DOCKER_IMAGE,
vol.Inclusive(ATTR_LAST_VERSION, 'custom_hass'): vol.Coerce(str),
vol.Optional(ATTR_PORT, default=8123): NETWORK_PORT,
vol.Optional(ATTR_PASSWORD): vol.Maybe(vol.Coerce(str)),
vol.Optional(ATTR_REFRESH_TOKEN): vol.Maybe(vol.Coerce(str)),
vol.Optional(ATTR_SSL, default=False): vol.Boolean(),
vol.Optional(ATTR_WATCHDOG, default=True): vol.Boolean(),
vol.Optional(ATTR_WAIT_BOOT, default=600):
vol.All(vol.Coerce(int), vol.Range(min=60)),
}, extra=vol.REMOVE_EXTRA)
SCHEMA_HASS_CONFIG = vol.Schema(
{
vol.Optional(ATTR_UUID, default=lambda: uuid.uuid4().hex): UUID_MATCH,
vol.Optional(ATTR_ACCESS_TOKEN): SHA256,
vol.Optional(ATTR_BOOT, default=True): vol.Boolean(),
vol.Inclusive(ATTR_IMAGE, "custom_hass"): DOCKER_IMAGE,
vol.Inclusive(ATTR_LAST_VERSION, "custom_hass"): vol.Coerce(str),
vol.Optional(ATTR_PORT, default=8123): NETWORK_PORT,
vol.Optional(ATTR_PASSWORD): vol.Maybe(vol.Coerce(str)),
vol.Optional(ATTR_REFRESH_TOKEN): vol.Maybe(vol.Coerce(str)),
vol.Optional(ATTR_SSL, default=False): vol.Boolean(),
vol.Optional(ATTR_WATCHDOG, default=True): vol.Boolean(),
vol.Optional(ATTR_WAIT_BOOT, default=600): vol.All(
vol.Coerce(int), vol.Range(min=60)
),
},
extra=vol.REMOVE_EXTRA,
)
SCHEMA_UPDATER_CONFIG = vol.Schema({
vol.Optional(ATTR_CHANNEL, default=CHANNEL_STABLE): CHANNELS,
vol.Optional(ATTR_HOMEASSISTANT): vol.Coerce(str),
vol.Optional(ATTR_HASSIO): vol.Coerce(str),
vol.Optional(ATTR_HASSOS): vol.Coerce(str),
vol.Optional(ATTR_HASSOS_CLI): vol.Coerce(str),
}, extra=vol.REMOVE_EXTRA)
SCHEMA_UPDATER_CONFIG = vol.Schema(
{
vol.Optional(ATTR_CHANNEL, default=CHANNEL_STABLE): CHANNELS,
vol.Optional(ATTR_HOMEASSISTANT): vol.Coerce(str),
vol.Optional(ATTR_HASSIO): vol.Coerce(str),
vol.Optional(ATTR_HASSOS): vol.Coerce(str),
vol.Optional(ATTR_HASSOS_CLI): vol.Coerce(str),
},
extra=vol.REMOVE_EXTRA,
)
# pylint: disable=no-value-for-parameter
SCHEMA_HASSIO_CONFIG = vol.Schema({
vol.Optional(ATTR_TIMEZONE, default='UTC'): validate_timezone,
vol.Optional(ATTR_LAST_BOOT): vol.Coerce(str),
vol.Optional(ATTR_ADDONS_CUSTOM_LIST, default=[
"https://github.com/hassio-addons/repository",
]): REPOSITORIES,
vol.Optional(ATTR_WAIT_BOOT, default=5): WAIT_BOOT,
}, extra=vol.REMOVE_EXTRA)
SCHEMA_HASSIO_CONFIG = vol.Schema(
{
vol.Optional(ATTR_TIMEZONE, default="UTC"): validate_timezone,
vol.Optional(ATTR_LAST_BOOT): vol.Coerce(str),
vol.Optional(
ATTR_ADDONS_CUSTOM_LIST,
default=["https://github.com/hassio-addons/repository"],
): REPOSITORIES,
vol.Optional(ATTR_WAIT_BOOT, default=5): WAIT_BOOT,
},
extra=vol.REMOVE_EXTRA,
)
SCHEMA_DISCOVERY = vol.Schema([
vol.Schema({
vol.Required(ATTR_UUID): UUID_MATCH,
vol.Required(ATTR_ADDON): vol.Coerce(str),
vol.Required(ATTR_SERVICE): SERVICE_ALL,
vol.Required(ATTR_CONFIG): vol.Maybe(dict),
}, extra=vol.REMOVE_EXTRA)
])
SCHEMA_DISCOVERY_CONFIG = vol.Schema({
vol.Optional(ATTR_DISCOVERY, default=list): schema_or(SCHEMA_DISCOVERY),
}, extra=vol.REMOVE_EXTRA)
SCHEMA_AUTH_CONFIG = vol.Schema({
SHA256: SHA256
})
SCHEMA_AUTH_CONFIG = vol.Schema({SHA256: SHA256})

View File

@ -0,0 +1 @@
"""Tests for discovery."""

View File

@ -0,0 +1,19 @@
"""Test DeConz discovery."""
import voluptuous as vol
import pytest
from hassio.discovery.validate import valid_discovery_config
def test_good_config():
"""Test good deconz config."""
valid_discovery_config("deconz", {"host": "test", "port": 3812})
def test_bad_config():
"""Test good deconz config."""
with pytest.raises(vol.Invalid):
valid_discovery_config("deconz", {"host": "test"})

View File

@ -0,0 +1,21 @@
"""Test MQTT discovery."""
import voluptuous as vol
import pytest
from hassio.discovery.validate import valid_discovery_config
def test_good_config():
"""Test good mqtt config."""
valid_discovery_config(
"mqtt", {"host": "test", "port": 3812, "username": "bla", "ssl": True}
)
def test_bad_config():
"""Test good mqtt config."""
with pytest.raises(vol.Invalid):
valid_discovery_config("mqtt", {"host": "test", "username": "bla", "ssl": True})

View File

@ -0,0 +1,21 @@
"""Test validate of discovery."""
import voluptuous as vol
import pytest
from hassio.discovery import validate
def test_valid_services():
"""Validate that service is valid."""
for service in ("mqtt", "deconz"):
validate.valid_discovery_service(service)
def test_invalid_services():
"""Test that validate is invalid for a service."""
for service in ("fadsfasd", "203432"):
with pytest.raises(vol.Invalid):
validate.valid_discovery_service(service)