105 lines
2.8 KiB
Python
105 lines
2.8 KiB
Python
|
"""Validate dependencies."""
|
||
|
|
import pathlib
|
||
|
|
from typing import Dict
|
||
|
|
|
||
|
|
import re
|
||
|
|
import voluptuous as vol
|
||
|
|
from voluptuous.humanize import humanize_error
|
||
|
|
|
||
|
|
from homeassistant.exceptions import HomeAssistantError
|
||
|
|
from homeassistant.helpers import config_validation as cv
|
||
|
|
from homeassistant.util.yaml import load_yaml
|
||
|
|
|
||
|
|
from .model import Integration
|
||
|
|
|
||
|
|
|
||
|
|
def exists(value):
|
||
|
|
"""Check if value exists."""
|
||
|
|
if value is None:
|
||
|
|
raise vol.Invalid("Value cannot be None")
|
||
|
|
return value
|
||
|
|
|
||
|
|
|
||
|
|
FIELD_SCHEMA = vol.Schema({
|
||
|
|
vol.Required('description'): str,
|
||
|
|
vol.Optional('example'): exists,
|
||
|
|
vol.Optional('default'): exists,
|
||
|
|
vol.Optional('values'): exists,
|
||
|
|
vol.Optional('required'): bool,
|
||
|
|
})
|
||
|
|
|
||
|
|
SERVICE_SCHEMA = vol.Schema({
|
||
|
|
vol.Required('description'): str,
|
||
|
|
vol.Optional('fields'): vol.Schema({
|
||
|
|
str: FIELD_SCHEMA
|
||
|
|
})
|
||
|
|
})
|
||
|
|
|
||
|
|
SERVICES_SCHEMA = vol.Schema({
|
||
|
|
cv.slug: SERVICE_SCHEMA
|
||
|
|
})
|
||
|
|
|
||
|
|
|
||
|
|
def grep_dir(path: pathlib.Path, glob_pattern: str, search_pattern: str) \
|
||
|
|
-> bool:
|
||
|
|
"""Recursively go through a dir and it's children and find the regex."""
|
||
|
|
pattern = re.compile(search_pattern)
|
||
|
|
|
||
|
|
for fil in path.glob(glob_pattern):
|
||
|
|
if not fil.is_file():
|
||
|
|
continue
|
||
|
|
|
||
|
|
if pattern.search(fil.read_text()):
|
||
|
|
return True
|
||
|
|
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def validate_services(integration: Integration):
|
||
|
|
"""Validate services."""
|
||
|
|
# Find if integration uses services
|
||
|
|
has_services = grep_dir(integration.path, "**/*.py",
|
||
|
|
r"hass\.(services|async_register)")
|
||
|
|
|
||
|
|
if not has_services:
|
||
|
|
return
|
||
|
|
|
||
|
|
try:
|
||
|
|
data = load_yaml(str(integration.path / 'services.yaml'))
|
||
|
|
except FileNotFoundError:
|
||
|
|
print(
|
||
|
|
"Warning: {} registeres services but has no services.yaml".format(
|
||
|
|
integration.domain))
|
||
|
|
# integration.add_error(
|
||
|
|
# 'services', 'Registers services but has no services.yaml')
|
||
|
|
return
|
||
|
|
except HomeAssistantError:
|
||
|
|
integration.add_error(
|
||
|
|
'services', 'Registers services but unable to load services.yaml')
|
||
|
|
return
|
||
|
|
|
||
|
|
try:
|
||
|
|
SERVICES_SCHEMA(data)
|
||
|
|
except vol.Invalid as err:
|
||
|
|
integration.add_error(
|
||
|
|
'services',
|
||
|
|
"Invalid services.yaml: {}".format(humanize_error(data, err)))
|
||
|
|
|
||
|
|
|
||
|
|
def validate(integrations: Dict[str, Integration], config):
|
||
|
|
"""Handle dependencies for integrations."""
|
||
|
|
# check services.yaml is cool
|
||
|
|
for integration in integrations.values():
|
||
|
|
if not integration.manifest:
|
||
|
|
continue
|
||
|
|
|
||
|
|
validate_services(integration)
|
||
|
|
|
||
|
|
# check that all referenced dependencies exist
|
||
|
|
for dep in integration.manifest['dependencies']:
|
||
|
|
if dep not in integrations:
|
||
|
|
integration.add_error(
|
||
|
|
'dependencies',
|
||
|
|
"Dependency {} does not exist"
|
||
|
|
)
|