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

Late review updates for Blink (#104755)

Browse source
This commit is contained in:
mkmer 2023-11-30 16:40:41 -05:00 • committed by Franck Nijhof
parent cc0326548e
commit 7ea4e15ff2
No known key found for this signature in database
GPG key ID: D62583BA8AB11CA3
4 changed files with 122 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
homeassistant/components/blink/__init__.py
View file

@ -25,7 +25,7 @@ from homeassistant.helpers.typing import ConfigType
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, PLATFORMS
from .coordinator import BlinkUpdateCoordinator
from .services import async_setup_services
from .services import setup_services
_LOGGER = logging.getLogger(__name__)
@ -74,7 +74,7 @@ async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Blink."""
await async_setup_services(hass)
setup_services(hass)
return True

79
homeassistant/components/blink/services.py
View file

@ -1,8 +1,6 @@
"""Services for the Blink integration."""
from __future__ import annotations
import logging
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
@ -14,7 +12,7 @@ from homeassistant.const import (
CONF_PIN,
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
import homeassistant.helpers.config_validation as cv
import homeassistant.helpers.device_registry as dr
@ -27,56 +25,67 @@ from .const import (
)
from .coordinator import BlinkUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
SERVICE_SAVE_VIDEO_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): cv.ensure_list,
vol.Required(ATTR_DEVICE_ID): vol.All(cv.ensure_list, [cv.string]),
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_FILENAME): cv.string,
}
)
SERVICE_SEND_PIN_SCHEMA = vol.Schema(
{vol.Required(ATTR_DEVICE_ID): cv.ensure_list, vol.Optional(CONF_PIN): cv.string}
{
vol.Required(ATTR_DEVICE_ID): vol.All(cv.ensure_list, [cv.string]),
vol.Optional(CONF_PIN): cv.string,
}
)
SERVICE_SAVE_RECENT_CLIPS_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): cv.ensure_list,
vol.Required(ATTR_DEVICE_ID): vol.All(cv.ensure_list, [cv.string]),
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_FILE_PATH): cv.string,
}
)
async def async_setup_services(hass: HomeAssistant) -> None:
def setup_services(hass: HomeAssistant) -> None:
"""Set up the services for the Blink integration."""
async def collect_coordinators(
def collect_coordinators(
device_ids: list[str],
) -> list[BlinkUpdateCoordinator]:
config_entries = list[ConfigEntry]()
config_entries: list[ConfigEntry] = []
registry = dr.async_get(hass)
for target in device_ids:
device = registry.async_get(target)
if device:
device_entries = list[ConfigEntry]()
device_entries: list[ConfigEntry] = []
for entry_id in device.config_entries:
entry = hass.config_entries.async_get_entry(entry_id)
if entry and entry.domain == DOMAIN:
device_entries.append(entry)
if not device_entries:
raise HomeAssistantError(
f"Device '{target}' is not a {DOMAIN} device"
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_device",
translation_placeholders={"target": target, "domain": DOMAIN},
)
config_entries.extend(device_entries)
else:
raise HomeAssistantError(
f"Device '{target}' not found in device registry"
translation_domain=DOMAIN,
translation_key="device_not_found",
translation_placeholders={"target": target},
)
coordinators = list[BlinkUpdateCoordinator]()
coordinators: list[BlinkUpdateCoordinator] = []
for config_entry in config_entries:
if config_entry.state != ConfigEntryState.LOADED:
raise HomeAssistantError(f"{config_entry.title} is not loaded")
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="not_loaded",
translation_placeholders={"target": config_entry.title},
)
coordinators.append(hass.data[DOMAIN][config_entry.entry_id])
return coordinators
@ -85,24 +94,36 @@ async def async_setup_services(hass: HomeAssistant) -> None:
camera_name = call.data[CONF_NAME]
video_path = call.data[CONF_FILENAME]
if not hass.config.is_allowed_path(video_path):
_LOGGER.error("Can't write %s, no access to path!", video_path)
return
for coordinator in await collect_coordinators(call.data[ATTR_DEVICE_ID]):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_path",
translation_placeholders={"target": video_path},
)
for coordinator in collect_coordinators(call.data[ATTR_DEVICE_ID]):
all_cameras = coordinator.api.cameras
if camera_name in all_cameras:
try:
await all_cameras[camera_name].video_to_file(video_path)
except OSError as err:
_LOGGER.error("Can't write image to file: %s", err)
raise ServiceValidationError(
str(err),
translation_domain=DOMAIN,
translation_key="cant_write",
) from err
async def async_handle_save_recent_clips_service(call: ServiceCall) -> None:
"""Save multiple recent clips to output directory."""
camera_name = call.data[CONF_NAME]
clips_dir = call.data[CONF_FILE_PATH]
if not hass.config.is_allowed_path(clips_dir):
_LOGGER.error("Can't write to directory %s, no access to path!", clips_dir)
return
for coordinator in await collect_coordinators(call.data[ATTR_DEVICE_ID]):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_path",
translation_placeholders={"target": clips_dir},
)
for coordinator in collect_coordinators(call.data[ATTR_DEVICE_ID]):
all_cameras = coordinator.api.cameras
if camera_name in all_cameras:
try:
@ -110,11 +131,15 @@ async def async_setup_services(hass: HomeAssistant) -> None:
output_dir=clips_dir
)
except OSError as err:
_LOGGER.error("Can't write recent clips to directory: %s", err)
raise ServiceValidationError(
str(err),
translation_domain=DOMAIN,
translation_key="cant_write",
) from err
async def send_pin(call: ServiceCall):
"""Call blink to send new pin."""
for coordinator in await collect_coordinators(call.data[ATTR_DEVICE_ID]):
for coordinator in collect_coordinators(call.data[ATTR_DEVICE_ID]):
await coordinator.api.auth.send_auth_key(
coordinator.api,
call.data[CONF_PIN],
@ -122,7 +147,7 @@ async def async_setup_services(hass: HomeAssistant) -> None:
async def blink_refresh(call: ServiceCall):
"""Call blink to refresh info."""
for coordinator in await collect_coordinators(call.data[ATTR_DEVICE_ID]):
for coordinator in collect_coordinators(call.data[ATTR_DEVICE_ID]):
await coordinator.api.refresh(force_cache=True)
# Register all the above services

17
homeassistant/components/blink/strings.json
View file

@ -101,5 +101,22 @@
}
}
}
},
"exceptions": {
"invalid_device": {
"message": "Device '{target}' is not a {domain} device"
},
"device_not_found": {
"message": "Device '{target}' not found in device registry"
},
"no_path": {
"message": "Can't write to directory {target}, no access to path!"
},
"cant_write": {
"message": "Can't write to file"
},
"not_loaded": {
"message": "{target} is not loaded"
}
}
}

38
tests/components/blink/test_services.py
View file

@ -19,7 +19,7 @@ from homeassistant.const import (
CONF_PIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import device_registry as dr
from tests.common import MockConfigEntry
@ -58,7 +58,7 @@ async def test_refresh_service_calls(
assert mock_blink_api.refresh.call_count == 2
with pytest.raises(HomeAssistantError) as execinfo:
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_REFRESH,
@ -66,8 +66,6 @@ async def test_refresh_service_calls(
blocking=True,
)
assert "Device 'bad-device_id' not found in device registry" in str(execinfo)
async def test_video_service_calls(
hass: HomeAssistant,
@ -90,7 +88,7 @@ async def test_video_service_calls(
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
caplog.clear()
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_VIDEO,
@ -101,7 +99,6 @@ async def test_video_service_calls(
},
blocking=True,
)
assert "no access to path!" in caplog.text
hass.config.is_allowed_path = Mock(return_value=True)
caplog.clear()
@ -118,7 +115,7 @@ async def test_video_service_calls(
)
mock_blink_api.cameras[CAMERA_NAME].video_to_file.assert_awaited_once()
with pytest.raises(HomeAssistantError) as execinfo:
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_VIDEO,
@ -130,11 +127,9 @@ async def test_video_service_calls(
blocking=True,
)
assert "Device 'bad-device_id' not found in device registry" in str(execinfo)
mock_blink_api.cameras[CAMERA_NAME].video_to_file = AsyncMock(side_effect=OSError)
caplog.clear()
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_VIDEO,
@ -145,7 +140,6 @@ async def test_video_service_calls(
},
blocking=True,
)
assert "Can't write image" in caplog.text
hass.config.is_allowed_path = Mock(return_value=False)
@ -171,7 +165,7 @@ async def test_picture_service_calls(
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
caplog.clear()
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
@ -182,7 +176,6 @@ async def test_picture_service_calls(
},
blocking=True,
)
assert "no access to path!" in caplog.text
hass.config.is_allowed_path = Mock(return_value=True)
mock_blink_api.cameras = {CAMERA_NAME: AsyncMock()}
@ -202,8 +195,8 @@ async def test_picture_service_calls(
mock_blink_api.cameras[CAMERA_NAME].save_recent_clips = AsyncMock(
side_effect=OSError
)
caplog.clear()
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
@ -214,9 +207,8 @@ async def test_picture_service_calls(
},
blocking=True,
)
assert "Can't write recent clips to directory" in caplog.text
with pytest.raises(HomeAssistantError) as execinfo:
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
@ -228,8 +220,6 @@ async def test_picture_service_calls(
blocking=True,
)
assert "Device 'bad-device_id' not found in device registry" in str(execinfo)
async def test_pin_service_calls(
hass: HomeAssistant,
@ -259,7 +249,7 @@ async def test_pin_service_calls(
)
assert mock_blink_api.auth.send_auth_key.assert_awaited_once
with pytest.raises(HomeAssistantError) as execinfo:
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_PIN,
@ -267,8 +257,6 @@ async def test_pin_service_calls(
blocking=True,
)
assert "Device 'bad-device_id' not found in device registry" in str(execinfo)
@pytest.mark.parametrize(
("service", "params"),
@ -325,7 +313,7 @@ async def test_service_called_with_non_blink_device(
parameters = {ATTR_DEVICE_ID: [device_entry.id]}
parameters.update(params)
with pytest.raises(HomeAssistantError) as execinfo:
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
service,
@ -333,8 +321,6 @@ async def test_service_called_with_non_blink_device(
blocking=True,
)
assert f"Device '{device_entry.id}' is not a blink device" in str(execinfo)
@pytest.mark.parametrize(
("service", "params"),
@ -382,12 +368,10 @@ async def test_service_called_with_unloaded_entry(
parameters = {ATTR_DEVICE_ID: [device_entry.id]}
parameters.update(params)
with pytest.raises(HomeAssistantError) as execinfo:
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
service,
parameters,
blocking=True,
)
assert "Mock Title is not loaded" in str(execinfo)