evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add save and delete WS commands to blueprints (#42907)

Browse source 
Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
Bram Kragten 2020-11-11 23:32:46 +01:00 • committed by GitHub
parent d47b3a5f44
commit 1c7080d5c5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 349 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
homeassistant/components/blueprint/const.py
View file

@ -5,5 +5,6 @@ CONF_BLUEPRINT = "blueprint"
CONF_USE_BLUEPRINT = "use_blueprint"
CONF_INPUT = "input"
CONF_SOURCE_URL = "source_url"
CONF_DESCRIPTION = "description"
DOMAIN = "blueprint"

8
homeassistant/components/blueprint/errors.py
View file

@ -78,3 +78,11 @@ class MissingPlaceholder(BlueprintWithNameException):
blueprint_name,
f"Missing placeholder {', '.join(sorted(placeholder_names))}",
)
class FileAlreadyExists(BlueprintWithNameException):
"""Error when file already exists."""
def __init__(self, domain: str, blueprint_name: str) -> None:
"""Initialize blueprint exception."""
super().__init__(domain, blueprint_name, "Blueprint already exists")

39
homeassistant/components/blueprint/models.py
View file

@ -24,6 +24,7 @@ from .const import (
from .errors import (
BlueprintException,
FailedToLoad,
FileAlreadyExists,
InvalidBlueprint,
InvalidBlueprintInputs,
MissingPlaceholder,
@ -86,6 +87,10 @@ class Blueprint:
if source_url is not None:
self.data[CONF_BLUEPRINT][CONF_SOURCE_URL] = source_url
def yaml(self) -> str:
"""Dump blueprint as YAML."""
return yaml.dump(self.data)
class BlueprintInputs:
"""Inputs for a blueprint."""
@ -229,3 +234,37 @@ class DomainBlueprints:
inputs = BlueprintInputs(blueprint, config_with_blueprint)
inputs.validate()
return inputs
async def async_remove_blueprint(self, blueprint_path: str) -> None:
"""Remove a blueprint file."""
path = pathlib.Path(
self.hass.config.path(BLUEPRINT_FOLDER, self.domain, blueprint_path)
)
await self.hass.async_add_executor_job(path.unlink)
self._blueprints[blueprint_path] = None
def _create_file(self, blueprint: Blueprint, blueprint_path: str) -> None:
"""Create blueprint file."""
path = pathlib.Path(
self.hass.config.path(BLUEPRINT_FOLDER, self.domain, blueprint_path)
)
if path.exists():
raise FileAlreadyExists(self.domain, blueprint_path)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(blueprint.yaml())
async def async_add_blueprint(
self, blueprint: Blueprint, blueprint_path: str
) -> None:
"""Add a blueprint."""
if not blueprint_path.endswith(".yaml"):
blueprint_path = f"{blueprint_path}.yaml"
await self.hass.async_add_executor_job(
self._create_file, blueprint, blueprint_path
)
self._blueprints[blueprint_path] = blueprint

24
homeassistant/components/blueprint/schemas.py
View file

@ -7,7 +7,13 @@ from homeassistant.const import CONF_DOMAIN, CONF_NAME, CONF_PATH
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
from .const import CONF_BLUEPRINT, CONF_INPUT, CONF_USE_BLUEPRINT
from .const import (
CONF_BLUEPRINT,
CONF_DESCRIPTION,
CONF_INPUT,
CONF_SOURCE_URL,
CONF_USE_BLUEPRINT,
)
@callback
@ -22,14 +28,26 @@ def is_blueprint_instance_config(config: Any) -> bool:
return isinstance(config, dict) and CONF_USE_BLUEPRINT in config
BLUEPRINT_INPUT_SCHEMA = vol.Schema(
{
vol.Optional(CONF_NAME): str,
vol.Optional(CONF_DESCRIPTION): str,
}
)
BLUEPRINT_SCHEMA = vol.Schema(
{
# No definition yet for the inputs.
vol.Required(CONF_BLUEPRINT): vol.Schema(
{
vol.Required(CONF_NAME): str,
vol.Required(CONF_DOMAIN): str,
vol.Optional(CONF_INPUT, default=dict): {str: None},
vol.Optional(CONF_SOURCE_URL): cv.url,
vol.Optional(CONF_INPUT, default=dict): {
str: vol.Any(
None,
BLUEPRINT_INPUT_SCHEMA,
)
},
}
),
},

116
homeassistant/components/blueprint/websocket_api.py
View file

@ -1,5 +1,4 @@
"""Websocket API for blueprint."""
import asyncio
import logging
from typing import Dict, Optional
@ -8,10 +7,13 @@ import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv
from homeassistant.util import yaml
from . import importer, models
from .const import DOMAIN
from .errors import FileAlreadyExists
_LOGGER = logging.getLogger(__package__)
@ -21,12 +23,15 @@ def async_setup(hass: HomeAssistant):
"""Set up the websocket API."""
websocket_api.async_register_command(hass, ws_list_blueprints)
websocket_api.async_register_command(hass, ws_import_blueprint)
websocket_api.async_register_command(hass, ws_save_blueprint)
websocket_api.async_register_command(hass, ws_delete_blueprint)
@websocket_api.async_response
@websocket_api.websocket_command(
{
vol.Required("type"): "blueprint/list",
vol.Required("domain"): cv.string,
}
)
async def ws_list_blueprints(hass, connection, msg):
@ -36,21 +41,19 @@ async def ws_list_blueprints(hass, connection, msg):
)
results = {}
for domain, domain_results in zip(
domain_blueprints,
await asyncio.gather(
*[db.async_get_blueprints() for db in domain_blueprints.values()]
),
):
for path, value in domain_results.items():
if isinstance(value, models.Blueprint):
domain_results[path] = {
"metadata": value.metadata,
}
else:
domain_results[path] = {"error": str(value)}
if msg["domain"] not in domain_blueprints:
connection.send_result(msg["id"], results)
return
results[domain] = domain_results
domain_results = await domain_blueprints[msg["domain"]].async_get_blueprints()
for path, value in domain_results.items():
if isinstance(value, models.Blueprint):
results[path] = {
"metadata": value.metadata,
}
else:
results[path] = {"error": str(value)}
connection.send_result(msg["id"], results)
@ -84,3 +87,86 @@ async def ws_import_blueprint(hass, connection, msg):
},
},
)
@websocket_api.async_response
@websocket_api.websocket_command(
{
vol.Required("type"): "blueprint/save",
vol.Required("domain"): cv.string,
vol.Required("path"): cv.path,
vol.Required("yaml"): cv.string,
vol.Optional("source_url"): cv.url,
}
)
async def ws_save_blueprint(hass, connection, msg):
"""Save a blueprint."""
path = msg["path"]
domain = msg["domain"]
domain_blueprints: Optional[Dict[str, models.DomainBlueprints]] = hass.data.get(
DOMAIN, {}
)
if domain not in domain_blueprints:
connection.send_error(
msg["id"], websocket_api.ERR_INVALID_FORMAT, "Unsupported domain"
)
try:
blueprint = models.Blueprint(
yaml.parse_yaml(msg["yaml"]), expected_domain=domain
)
if "source_url" in msg:
blueprint.update_metadata(source_url=msg["source_url"])
except HomeAssistantError as err:
connection.send_error(msg["id"], websocket_api.ERR_INVALID_FORMAT, str(err))
return
try:
await domain_blueprints[domain].async_add_blueprint(blueprint, path)
except FileAlreadyExists:
connection.send_error(msg["id"], "already_exists", "File already exists")
return
except OSError as err:
connection.send_error(msg["id"], websocket_api.ERR_UNKNOWN_ERROR, str(err))
return
connection.send_result(
msg["id"],
)
@websocket_api.async_response
@websocket_api.websocket_command(
{
vol.Required("type"): "blueprint/delete",
vol.Required("domain"): cv.string,
vol.Required("path"): cv.path,
}
)
async def ws_delete_blueprint(hass, connection, msg):
"""Delete a blueprint."""
path = msg["path"]
domain = msg["domain"]
domain_blueprints: Optional[Dict[str, models.DomainBlueprints]] = hass.data.get(
DOMAIN, {}
)
if domain not in domain_blueprints:
connection.send_error(
msg["id"], websocket_api.ERR_INVALID_FORMAT, "Unsupported domain"
)
try:
await domain_blueprints[domain].async_remove_blueprint(path)
except OSError as err:
connection.send_error(msg["id"], websocket_api.ERR_UNKNOWN_ERROR, str(err))
return
connection.send_result(
msg["id"],
)

24
tests/components/blueprint/test_models.py
View file

@ -17,7 +17,10 @@ def blueprint_1():
"blueprint": {
"name": "Hello",
"domain": "automation",
"input": {"test-placeholder": None},
"source_url": "https://github.com/balloob/home-assistant-config/blob/main/blueprints/automation/motion_light.yaml",
"input": {
"test-placeholder": {"name": "Name", "description": "Description"}
},
},
"example": Placeholder("test-placeholder"),
}
@ -59,7 +62,8 @@ def test_blueprint_properties(blueprint_1):
assert blueprint_1.metadata == {
"name": "Hello",
"domain": "automation",
"input": {"test-placeholder": None},
"source_url": "https://github.com/balloob/home-assistant-config/blob/main/blueprints/automation/motion_light.yaml",
"input": {"test-placeholder": {"name": "Name", "description": "Description"}},
}
assert blueprint_1.domain == "automation"
assert blueprint_1.name == "Hello"
@ -152,3 +156,19 @@ async def test_domain_blueprints_inputs_from_config(domain_bps, blueprint_1):
)
assert inputs.blueprint is blueprint_1
assert inputs.inputs == {"test-placeholder": None}
async def test_domain_blueprints_add_blueprint(domain_bps, blueprint_1):
"""Test DomainBlueprints.async_add_blueprint."""
with patch.object(domain_bps, "_create_file") as create_file_mock:
# Should add extension when not present.
await domain_bps.async_add_blueprint(blueprint_1, "something")
assert create_file_mock.call_args[0][1] == ("something.yaml")
await domain_bps.async_add_blueprint(blueprint_1, "something2.yaml")
assert create_file_mock.call_args[0][1] == ("something2.yaml")
# Should be in cache.
with patch.object(domain_bps, "_load_blueprint") as mock_load:
assert await domain_bps.async_get_blueprint("something.yaml") == blueprint_1
assert not mock_load.mock_calls

160
tests/components/blueprint/test_websocket_api.py
View file

@ -6,6 +6,8 @@ import pytest
from homeassistant.components import automation
from homeassistant.setup import async_setup_component
from tests.async_mock import Mock, patch
@pytest.fixture(autouse=True)
async def setup_bp(hass):
@ -19,14 +21,14 @@ async def setup_bp(hass):
async def test_list_blueprints(hass, hass_ws_client):
"""Test listing blueprints."""
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "blueprint/list"})
await client.send_json({"id": 5, "type": "blueprint/list", "domain": "automation"})
msg = await client.receive_json()
assert msg["id"] == 5
assert msg["success"]
blueprints = msg["result"]
assert blueprints.get("automation") == {
assert blueprints == {
"test_event_service.yaml": {
"metadata": {
"domain": "automation",
@ -44,8 +46,23 @@ async def test_list_blueprints(hass, hass_ws_client):
}
async def test_import_blueprint(hass, aioclient_mock, hass_ws_client):
async def test_list_blueprints_non_existing_domain(hass, hass_ws_client):
"""Test listing blueprints."""
client = await hass_ws_client(hass)
await client.send_json(
{"id": 5, "type": "blueprint/list", "domain": "not_existsing"}
)
msg = await client.receive_json()
assert msg["id"] == 5
assert msg["success"]
blueprints = msg["result"]
assert blueprints == {}
async def test_import_blueprint(hass, aioclient_mock, hass_ws_client):
"""Test importing blueprints."""
raw_data = Path(
hass.config.path("blueprints/automation/test_event_service.yaml")
).read_text()
@ -80,3 +97,140 @@ async def test_import_blueprint(hass, aioclient_mock, hass_ws_client):
},
},
}
async def test_save_blueprint(hass, aioclient_mock, hass_ws_client):
"""Test saving blueprints."""
raw_data = Path(
hass.config.path("blueprints/automation/test_event_service.yaml")
).read_text()
with patch("pathlib.Path.write_text") as write_mock:
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 6,
"type": "blueprint/save",
"path": "test_save",
"yaml": raw_data,
"domain": "automation",
"source_url": "https://github.com/balloob/home-assistant-config/blob/main/blueprints/automation/motion_light.yaml",
}
)
msg = await client.receive_json()
assert msg["id"] == 6
assert msg["success"]
assert write_mock.mock_calls
assert write_mock.call_args[0] == (
"blueprint:\n name: Call service based on event\n domain: automation\n input:\n trigger_event:\n service_to_call:\n source_url: https://github.com/balloob/home-assistant-config/blob/main/blueprints/automation/motion_light.yaml\ntrigger:\n platform: event\n event_type: !placeholder 'trigger_event'\naction:\n service: !placeholder 'service_to_call'\n",
)
async def test_save_existing_file(hass, aioclient_mock, hass_ws_client):
"""Test saving blueprints."""
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 7,
"type": "blueprint/save",
"path": "test_event_service",
"yaml": 'blueprint: {name: "name", domain: "automation"}',
"domain": "automation",
"source_url": "https://github.com/balloob/home-assistant-config/blob/main/blueprints/automation/motion_light.yaml",
}
)
msg = await client.receive_json()
assert msg["id"] == 7
assert not msg["success"]
assert msg["error"] == {"code": "already_exists", "message": "File already exists"}
async def test_save_file_error(hass, aioclient_mock, hass_ws_client):
"""Test saving blueprints with OS error."""
with patch("pathlib.Path.write_text", side_effect=OSError):
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 8,
"type": "blueprint/save",
"path": "test_save",
"yaml": "raw_data",
"domain": "automation",
"source_url": "https://github.com/balloob/home-assistant-config/blob/main/blueprints/automation/motion_light.yaml",
}
)
msg = await client.receive_json()
assert msg["id"] == 8
assert not msg["success"]
async def test_save_invalid_blueprint(hass, aioclient_mock, hass_ws_client):
"""Test saving invalid blueprints."""
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 8,
"type": "blueprint/save",
"path": "test_wrong",
"yaml": "wrong_blueprint",
"domain": "automation",
"source_url": "https://github.com/balloob/home-assistant-config/blob/main/blueprints/automation/motion_light.yaml",
}
)
msg = await client.receive_json()
assert msg["id"] == 8
assert not msg["success"]
assert msg["error"] == {
"code": "invalid_format",
"message": "Invalid blueprint: expected a dictionary. Got 'wrong_blueprint'",
}
async def test_delete_blueprint(hass, aioclient_mock, hass_ws_client):
"""Test deleting blueprints."""
with patch("pathlib.Path.unlink", return_value=Mock()) as unlink_mock:
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 9,
"type": "blueprint/delete",
"path": "test_delete",
"domain": "automation",
}
)
msg = await client.receive_json()
assert unlink_mock.mock_calls
assert msg["id"] == 9
assert msg["success"]
async def test_delete_non_exist_file_blueprint(hass, aioclient_mock, hass_ws_client):
"""Test deleting non existing blueprints."""
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 9,
"type": "blueprint/delete",
"path": "none_existing",
"domain": "automation",
}
)
msg = await client.receive_json()
assert msg["id"] == 9
assert not msg["success"]