Move to new service handeling methods in Blink (#103435)

This commit is contained in:
mkmer 2023-11-28 02:34:41 -05:00 committed by GitHub
parent a1aff5f4a0
commit d318155f09
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 561 additions and 256 deletions

View file

@ -115,6 +115,7 @@ omit =
homeassistant/components/beewi_smartclim/sensor.py
homeassistant/components/bitcoin/sensor.py
homeassistant/components/bizkaibus/sensor.py
homeassistant/components/blink/__init__.py
homeassistant/components/blink/alarm_control_panel.py
homeassistant/components/blink/binary_sensor.py
homeassistant/components/blink/camera.py

View file

@ -21,17 +21,11 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import (
DEFAULT_SCAN_INTERVAL,
DOMAIN,
PLATFORMS,
SERVICE_REFRESH,
SERVICE_SAVE_RECENT_CLIPS,
SERVICE_SAVE_VIDEO,
SERVICE_SEND_PIN,
)
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, PLATFORMS
from .coordinator import BlinkUpdateCoordinator
from .services import async_setup_services
_LOGGER = logging.getLogger(__name__)
@ -43,6 +37,8 @@ SERVICE_SAVE_RECENT_CLIPS_SCHEMA = vol.Schema(
{vol.Required(CONF_NAME): cv.string, vol.Required(CONF_FILE_PATH): cv.string}
)
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def _reauth_flow_wrapper(hass, data):
"""Reauth flow wrapper."""
@ -75,6 +71,14 @@ async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Blink."""
await async_setup_services(hass)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Blink via config entry."""
hass.data.setdefault(DOMAIN, {})
@ -105,40 +109,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))
async def blink_refresh(event_time=None):
"""Call blink to refresh info."""
await coordinator.api.refresh(force_cache=True)
async def async_save_video(call):
"""Call save video service handler."""
await async_handle_save_video_service(hass, entry, call)
async def async_save_recent_clips(call):
"""Call save recent clips service handler."""
await async_handle_save_recent_clips_service(hass, entry, call)
async def send_pin(call):
"""Call blink to send new pin."""
pin = call.data[CONF_PIN]
await coordinator.api.auth.send_auth_key(
hass.data[DOMAIN][entry.entry_id].api,
pin,
)
hass.services.async_register(DOMAIN, SERVICE_REFRESH, blink_refresh)
hass.services.async_register(
DOMAIN, SERVICE_SAVE_VIDEO, async_save_video, schema=SERVICE_SAVE_VIDEO_SCHEMA
)
hass.services.async_register(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
async_save_recent_clips,
schema=SERVICE_SAVE_RECENT_CLIPS_SCHEMA,
)
hass.services.async_register(
DOMAIN, SERVICE_SEND_PIN, send_pin, schema=SERVICE_SEND_PIN_SCHEMA
)
return True
@ -158,13 +128,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload Blink entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
if len(hass.data[DOMAIN]) > 0:
return unload_ok
hass.services.async_remove(DOMAIN, SERVICE_REFRESH)
hass.services.async_remove(DOMAIN, SERVICE_SAVE_VIDEO)
hass.services.async_remove(DOMAIN, SERVICE_SEND_PIN)
return unload_ok
@ -172,37 +135,3 @@ async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
blink: Blink = hass.data[DOMAIN][entry.entry_id].api
blink.refresh_rate = entry.options[CONF_SCAN_INTERVAL]
async def async_handle_save_video_service(
hass: HomeAssistant, entry: ConfigEntry, call
) -> None:
"""Handle save video service calls."""
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
all_cameras = hass.data[DOMAIN][entry.entry_id].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)
async def async_handle_save_recent_clips_service(
hass: HomeAssistant, entry: ConfigEntry, call
) -> 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
all_cameras = hass.data[DOMAIN][entry.entry_id].api.cameras
if camera_name in all_cameras:
try:
await all_cameras[camera_name].save_recent_clips(output_dir=clips_dir)
except OSError as err:
_LOGGER.error("Can't write recent clips to directory: %s", err)

View file

@ -13,6 +13,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = 30
class BlinkUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
@ -25,7 +26,7 @@ class BlinkUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=30),
update_interval=timedelta(seconds=SCAN_INTERVAL),
)
async def _async_update_data(self) -> dict[str, Any]:

View file

@ -0,0 +1,150 @@
"""Services for the Blink integration."""
from __future__ import annotations
import logging
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import (
ATTR_DEVICE_ID,
CONF_FILE_PATH,
CONF_FILENAME,
CONF_NAME,
CONF_PIN,
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import HomeAssistantError
import homeassistant.helpers.config_validation as cv
import homeassistant.helpers.device_registry as dr
from .const import (
DOMAIN,
SERVICE_REFRESH,
SERVICE_SAVE_RECENT_CLIPS,
SERVICE_SAVE_VIDEO,
SERVICE_SEND_PIN,
)
from .coordinator import BlinkUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
SERVICE_SAVE_VIDEO_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): cv.ensure_list,
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}
)
SERVICE_SAVE_RECENT_CLIPS_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): cv.ensure_list,
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_FILE_PATH): cv.string,
}
)
async def async_setup_services(hass: HomeAssistant) -> None:
"""Set up the services for the Blink integration."""
async def collect_coordinators(
device_ids: list[str],
) -> list[BlinkUpdateCoordinator]:
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]()
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"
)
config_entries.extend(device_entries)
else:
raise HomeAssistantError(
f"Device '{target}' not found in device registry"
)
coordinators = list[BlinkUpdateCoordinator]()
for config_entry in config_entries:
if config_entry.state != ConfigEntryState.LOADED:
raise HomeAssistantError(f"{config_entry.title} is not loaded")
coordinators.append(hass.data[DOMAIN][config_entry.entry_id])
return coordinators
async def async_handle_save_video_service(call: ServiceCall) -> None:
"""Handle save video service calls."""
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]):
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)
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]):
all_cameras = coordinator.api.cameras
if camera_name in all_cameras:
try:
await all_cameras[camera_name].save_recent_clips(
output_dir=clips_dir
)
except OSError as err:
_LOGGER.error("Can't write recent clips to directory: %s", err)
async def send_pin(call: ServiceCall):
"""Call blink to send new pin."""
for coordinator in await collect_coordinators(call.data[ATTR_DEVICE_ID]):
await coordinator.api.auth.send_auth_key(
coordinator.api,
call.data[CONF_PIN],
)
async def blink_refresh(call: ServiceCall):
"""Call blink to refresh info."""
for coordinator in await collect_coordinators(call.data[ATTR_DEVICE_ID]):
await coordinator.api.refresh(force_cache=True)
# Register all the above services
service_mapping = [
(blink_refresh, SERVICE_REFRESH, None),
(
async_handle_save_video_service,
SERVICE_SAVE_VIDEO,
SERVICE_SAVE_VIDEO_SCHEMA,
),
(
async_handle_save_recent_clips_service,
SERVICE_SAVE_RECENT_CLIPS,
SERVICE_SAVE_RECENT_CLIPS_SCHEMA,
),
(send_pin, SERVICE_SEND_PIN, SERVICE_SEND_PIN_SCHEMA),
]
for service_handler, service_name, schema in service_mapping:
hass.services.async_register(
DOMAIN,
service_name,
service_handler,
schema=schema,
)

View file

@ -13,7 +13,7 @@ from tests.common import MockConfigEntry
CAMERA_ATTRIBUTES = {
"name": "Camera 1",
"camera_id": "111111",
"serial": "serail",
"serial": "serial",
"temperature": None,
"temperature_c": 25.1,
"temperature_calibrated": None,

View file

@ -1,6 +1,6 @@
"""Test the Blink init."""
import asyncio
from unittest.mock import AsyncMock, MagicMock, Mock
from unittest.mock import AsyncMock, MagicMock
from aiohttp import ClientError
import pytest
@ -8,12 +8,10 @@ import pytest
from homeassistant.components.blink.const import (
DOMAIN,
SERVICE_REFRESH,
SERVICE_SAVE_RECENT_CLIPS,
SERVICE_SAVE_VIDEO,
SERVICE_SEND_PIN,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_FILE_PATH, CONF_FILENAME, CONF_NAME, CONF_PIN
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@ -61,25 +59,6 @@ async def test_setup_not_ready_authkey_required(
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
async def test_unload_entry(
hass: HomeAssistant,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test being able to unload an entry."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
assert not hass.services.has_service(DOMAIN, SERVICE_REFRESH)
assert not hass.services.has_service(DOMAIN, SERVICE_SAVE_VIDEO)
assert not hass.services.has_service(DOMAIN, SERVICE_SEND_PIN)
async def test_unload_entry_multiple(
hass: HomeAssistant,
mock_blink_api: MagicMock,
@ -135,151 +114,3 @@ async def test_migrate(
await hass.async_block_till_done()
entry = hass.config_entries.async_get_entry(mock_config_entry.entry_id)
assert entry.state is ConfigEntryState.MIGRATION_ERROR
async def test_refresh_service_calls(
hass: HomeAssistant,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test refrest service calls."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
await hass.services.async_call(
DOMAIN,
SERVICE_REFRESH,
blocking=True,
)
assert mock_blink_api.refresh.call_count == 2
async def test_video_service_calls(
hass: HomeAssistant,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test video service calls."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
caplog.clear()
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_VIDEO,
{CONF_NAME: CAMERA_NAME, CONF_FILENAME: FILENAME},
blocking=True,
)
assert "no access to path!" in caplog.text
hass.config.is_allowed_path = Mock(return_value=True)
caplog.clear()
mock_blink_api.cameras = {CAMERA_NAME: AsyncMock()}
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_VIDEO,
{CONF_NAME: CAMERA_NAME, CONF_FILENAME: FILENAME},
blocking=True,
)
mock_blink_api.cameras[CAMERA_NAME].video_to_file.assert_awaited_once()
mock_blink_api.cameras[CAMERA_NAME].video_to_file = AsyncMock(side_effect=OSError)
caplog.clear()
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_VIDEO,
{CONF_NAME: CAMERA_NAME, CONF_FILENAME: FILENAME},
blocking=True,
)
assert "Can't write image" in caplog.text
hass.config.is_allowed_path = Mock(return_value=False)
async def test_picture_service_calls(
hass: HomeAssistant,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test picture servcie calls."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
caplog.clear()
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
{CONF_NAME: CAMERA_NAME, CONF_FILE_PATH: FILENAME},
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()}
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
{CONF_NAME: CAMERA_NAME, CONF_FILE_PATH: FILENAME},
blocking=True,
)
mock_blink_api.cameras[CAMERA_NAME].save_recent_clips.assert_awaited_once()
mock_blink_api.cameras[CAMERA_NAME].save_recent_clips = AsyncMock(
side_effect=OSError
)
caplog.clear()
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
{CONF_NAME: CAMERA_NAME, CONF_FILE_PATH: FILENAME},
blocking=True,
)
assert "Can't write recent clips to directory" in caplog.text
async def test_pin_service_calls(
hass: HomeAssistant,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test pin service calls."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_PIN,
{CONF_PIN: PIN},
blocking=True,
)
assert mock_blink_api.auth.send_auth_key.assert_awaited_once

View file

@ -0,0 +1,393 @@
"""Test the Blink services."""
from unittest.mock import AsyncMock, MagicMock, Mock
import pytest
from homeassistant.components.blink.const import (
DOMAIN,
SERVICE_REFRESH,
SERVICE_SAVE_RECENT_CLIPS,
SERVICE_SAVE_VIDEO,
SERVICE_SEND_PIN,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
ATTR_DEVICE_ID,
CONF_FILE_PATH,
CONF_FILENAME,
CONF_NAME,
CONF_PIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from tests.common import MockConfigEntry
CAMERA_NAME = "Camera 1"
FILENAME = "blah"
PIN = "1234"
async def test_refresh_service_calls(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test refrest service calls."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "12345")})
assert device_entry
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
await hass.services.async_call(
DOMAIN,
SERVICE_REFRESH,
{ATTR_DEVICE_ID: [device_entry.id]},
blocking=True,
)
assert mock_blink_api.refresh.call_count == 2
with pytest.raises(HomeAssistantError) as execinfo:
await hass.services.async_call(
DOMAIN,
SERVICE_REFRESH,
{ATTR_DEVICE_ID: ["bad-device_id"]},
blocking=True,
)
assert "Device 'bad-device_id' not found in device registry" in str(execinfo)
async def test_video_service_calls(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test video service calls."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "12345")})
assert device_entry
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
caplog.clear()
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_VIDEO,
{
ATTR_DEVICE_ID: [device_entry.id],
CONF_NAME: CAMERA_NAME,
CONF_FILENAME: FILENAME,
},
blocking=True,
)
assert "no access to path!" in caplog.text
hass.config.is_allowed_path = Mock(return_value=True)
caplog.clear()
mock_blink_api.cameras = {CAMERA_NAME: AsyncMock()}
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_VIDEO,
{
ATTR_DEVICE_ID: [device_entry.id],
CONF_NAME: CAMERA_NAME,
CONF_FILENAME: FILENAME,
},
blocking=True,
)
mock_blink_api.cameras[CAMERA_NAME].video_to_file.assert_awaited_once()
with pytest.raises(HomeAssistantError) as execinfo:
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_VIDEO,
{
ATTR_DEVICE_ID: ["bad-device_id"],
CONF_NAME: CAMERA_NAME,
CONF_FILENAME: FILENAME,
},
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()
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_VIDEO,
{
ATTR_DEVICE_ID: [device_entry.id],
CONF_NAME: CAMERA_NAME,
CONF_FILENAME: FILENAME,
},
blocking=True,
)
assert "Can't write image" in caplog.text
hass.config.is_allowed_path = Mock(return_value=False)
async def test_picture_service_calls(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test picture servcie calls."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "12345")})
assert device_entry
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
caplog.clear()
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
{
ATTR_DEVICE_ID: [device_entry.id],
CONF_NAME: CAMERA_NAME,
CONF_FILE_PATH: FILENAME,
},
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()}
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
{
ATTR_DEVICE_ID: [device_entry.id],
CONF_NAME: CAMERA_NAME,
CONF_FILE_PATH: FILENAME,
},
blocking=True,
)
mock_blink_api.cameras[CAMERA_NAME].save_recent_clips.assert_awaited_once()
mock_blink_api.cameras[CAMERA_NAME].save_recent_clips = AsyncMock(
side_effect=OSError
)
caplog.clear()
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
{
ATTR_DEVICE_ID: [device_entry.id],
CONF_NAME: CAMERA_NAME,
CONF_FILE_PATH: FILENAME,
},
blocking=True,
)
assert "Can't write recent clips to directory" in caplog.text
with pytest.raises(HomeAssistantError) as execinfo:
await hass.services.async_call(
DOMAIN,
SERVICE_SAVE_RECENT_CLIPS,
{
ATTR_DEVICE_ID: ["bad-device_id"],
CONF_NAME: CAMERA_NAME,
CONF_FILE_PATH: FILENAME,
},
blocking=True,
)
assert "Device 'bad-device_id' not found in device registry" in str(execinfo)
async def test_pin_service_calls(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test pin service calls."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "12345")})
assert device_entry
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_PIN,
{ATTR_DEVICE_ID: [device_entry.id], CONF_PIN: PIN},
blocking=True,
)
assert mock_blink_api.auth.send_auth_key.assert_awaited_once
with pytest.raises(HomeAssistantError) as execinfo:
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_PIN,
{ATTR_DEVICE_ID: ["bad-device_id"], CONF_PIN: PIN},
blocking=True,
)
assert "Device 'bad-device_id' not found in device registry" in str(execinfo)
@pytest.mark.parametrize(
("service", "params"),
[
(SERVICE_SEND_PIN, {CONF_PIN: PIN}),
(
SERVICE_SAVE_RECENT_CLIPS,
{
CONF_NAME: CAMERA_NAME,
CONF_FILE_PATH: FILENAME,
},
),
(
SERVICE_SAVE_VIDEO,
{
CONF_NAME: CAMERA_NAME,
CONF_FILENAME: FILENAME,
},
),
],
)
async def test_service_called_with_non_blink_device(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
service,
params,
) -> None:
"""Test service calls with non blink device."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
other_domain = "NotBlink"
other_config_id = "555"
await hass.config_entries.async_add(
MockConfigEntry(
title="Not Blink", domain=other_domain, entry_id=other_config_id
)
)
device_entry = device_registry.async_get_or_create(
config_entry_id=other_config_id,
identifiers={
(other_domain, 1),
},
)
hass.config.is_allowed_path = Mock(return_value=True)
mock_blink_api.cameras = {CAMERA_NAME: AsyncMock()}
parameters = {ATTR_DEVICE_ID: [device_entry.id]}
parameters.update(params)
with pytest.raises(HomeAssistantError) as execinfo:
await hass.services.async_call(
DOMAIN,
service,
parameters,
blocking=True,
)
assert f"Device '{device_entry.id}' is not a blink device" in str(execinfo)
@pytest.mark.parametrize(
("service", "params"),
[
(SERVICE_SEND_PIN, {CONF_PIN: PIN}),
(
SERVICE_SAVE_RECENT_CLIPS,
{
CONF_NAME: CAMERA_NAME,
CONF_FILE_PATH: FILENAME,
},
),
(
SERVICE_SAVE_VIDEO,
{
CONF_NAME: CAMERA_NAME,
CONF_FILENAME: FILENAME,
},
),
],
)
async def test_service_called_with_unloaded_entry(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
service,
params,
) -> None:
"""Test service calls with unloaded config entry."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await mock_config_entry.async_unload(hass)
device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "12345")})
assert device_entry
hass.config.is_allowed_path = Mock(return_value=True)
mock_blink_api.cameras = {CAMERA_NAME: AsyncMock()}
parameters = {ATTR_DEVICE_ID: [device_entry.id]}
parameters.update(params)
with pytest.raises(HomeAssistantError) as execinfo:
await hass.services.async_call(
DOMAIN,
service,
parameters,
blocking=True,
)
assert "Mock Title is not loaded" in str(execinfo)