jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-11-16 14:30:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Handle new Blink login flow (#154632)

Browse Source 
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Kira
2025-10-28 21:43:32 +01:00
committed by GitHub
parent 71589d212f
commit 1a1f3d6b4e
15 changed files with 320 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
homeassistant/components/blink/__init__.py
View File

@@ -4,7 +4,6 @@ from copy import deepcopy
import logging
from typing import Any
from aiohttp import ClientError
from blinkpy.auth import Auth
from blinkpy.blinkpy import Blink
import voluptuous as vol
@@ -18,7 +17,6 @@ from homeassistant.const import (
CONF_SCAN_INTERVAL,
)
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
@@ -83,22 +81,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: BlinkConfigEntry) -> boo
session = async_get_clientsession(hass)
blink = Blink(session=session)
auth_data = deepcopy(dict(entry.data))
blink.auth = Auth(auth_data, no_prompt=True, session=session)
blink.auth = Auth(
auth_data,
no_prompt=True,
session=session,
callback=lambda: _async_update_entry_data(hass, entry, blink),
)
blink.refresh_rate = entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
coordinator = BlinkUpdateCoordinator(hass, entry, blink)
try:
await blink.start()
except (ClientError, TimeoutError) as ex:
raise ConfigEntryNotReady("Can not connect to host") from ex
if blink.auth.check_key_required():
_LOGGER.debug("Attempting a reauth flow")
raise ConfigEntryAuthFailed("Need 2FA for Blink")
if not blink.available:
raise ConfigEntryNotReady
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
@@ -108,6 +99,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: BlinkConfigEntry) -> boo
return True
@callback
def _async_update_entry_data(
hass: HomeAssistant, entry: BlinkConfigEntry, blink: Blink
) -> None:
"""Update the config entry data after token refresh."""
hass.config_entries.async_update_entry(entry, data=blink.auth.login_attributes)
@callback
def _async_import_options_from_data_if_missing(
hass: HomeAssistant, entry: BlinkConfigEntry

9
homeassistant/components/blink/alarm_control_panel.py
View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import logging
from blinkpy.auth import UnauthorizedError
from blinkpy.blinkpy import Blink, BlinkSyncModule
from homeassistant.components.alarm_control_panel import (
@@ -13,7 +14,7 @@ from homeassistant.components.alarm_control_panel import (
)
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
@@ -91,6 +92,9 @@ class BlinkSyncModuleHA(
except TimeoutError as er:
raise HomeAssistantError("Blink failed to disarm camera") from er
except UnauthorizedError as er:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise ConfigEntryAuthFailed("Blink authorization failed") from er
await self.coordinator.async_refresh()
@@ -101,5 +105,8 @@ class BlinkSyncModuleHA(
except TimeoutError as er:
raise HomeAssistantError("Blink failed to arm camera away") from er
except UnauthorizedError as er:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise ConfigEntryAuthFailed("Blink authorization failed") from er
await self.coordinator.async_refresh()

30
homeassistant/components/blink/camera.py
View File

@@ -6,13 +6,19 @@ from collections.abc import Mapping
import logging
from typing import Any
from blinkpy.auth import UnauthorizedError
from blinkpy.camera import BlinkCamera as BlinkCameraAPI
from requests.exceptions import ChunkedEncodingError
import voluptuous as vol
from homeassistant.components.camera import Camera
from homeassistant.const import CONF_FILE_PATH, CONF_FILENAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
HomeAssistantError,
ServiceValidationError,
)
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
@@ -71,7 +77,9 @@ class BlinkCamera(CoordinatorEntity[BlinkUpdateCoordinator], Camera):
_attr_has_entity_name = True
_attr_name = None
def __init__(self, coordinator: BlinkUpdateCoordinator, name, camera) -> None:
def __init__(
self, coordinator: BlinkUpdateCoordinator, name, camera: BlinkCameraAPI
) -> None:
"""Initialize a camera."""
super().__init__(coordinator)
Camera.__init__(self)
@@ -101,6 +109,9 @@ class BlinkCamera(CoordinatorEntity[BlinkUpdateCoordinator], Camera):
translation_domain=DOMAIN,
translation_key="failed_arm",
) from er
except UnauthorizedError as er:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise ConfigEntryAuthFailed("Blink authorization failed") from er
self._camera.motion_enabled = True
await self.coordinator.async_refresh()
@@ -114,6 +125,9 @@ class BlinkCamera(CoordinatorEntity[BlinkUpdateCoordinator], Camera):
translation_domain=DOMAIN,
translation_key="failed_disarm",
) from er
except UnauthorizedError as er:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise ConfigEntryAuthFailed("Blink authorization failed") from er
self._camera.motion_enabled = False
await self.coordinator.async_refresh()
@@ -137,6 +151,9 @@ class BlinkCamera(CoordinatorEntity[BlinkUpdateCoordinator], Camera):
translation_domain=DOMAIN,
translation_key="failed_clip",
) from er
except UnauthorizedError as er:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise ConfigEntryAuthFailed("Blink authorization failed") from er
self.async_write_ha_state()
@@ -149,6 +166,9 @@ class BlinkCamera(CoordinatorEntity[BlinkUpdateCoordinator], Camera):
translation_domain=DOMAIN,
translation_key="failed_snap",
) from er
except UnauthorizedError as er:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise ConfigEntryAuthFailed("Blink authorization failed") from er
self.async_write_ha_state()
@@ -182,6 +202,9 @@ class BlinkCamera(CoordinatorEntity[BlinkUpdateCoordinator], Camera):
translation_domain=DOMAIN,
translation_key="cant_write",
) from err
except UnauthorizedError as er:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise ConfigEntryAuthFailed("Blink authorization failed") from er
async def save_video(self, filename) -> None:
"""Handle save video service calls."""
@@ -200,3 +223,6 @@ class BlinkCamera(CoordinatorEntity[BlinkUpdateCoordinator], Camera):
translation_domain=DOMAIN,
translation_key="cant_write",
) from err
except UnauthorizedError as er:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise ConfigEntryAuthFailed("Blink authorization failed") from er

148
homeassistant/components/blink/config_flow.py
View File

@@ -6,13 +6,18 @@ from collections.abc import Mapping
import logging
from typing import Any
from blinkpy.auth import Auth, LoginError, TokenRefreshFailed
from blinkpy.auth import Auth, BlinkTwoFARequiredError, LoginError, TokenRefreshFailed
from blinkpy.blinkpy import Blink, BlinkSetupError
import voluptuous as vol
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import (
SOURCE_REAUTH,
SOURCE_RECONFIGURE,
ConfigFlow,
ConfigFlowResult,
)
from homeassistant.const import CONF_PASSWORD, CONF_PIN, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@@ -21,23 +26,18 @@ from .const import DEVICE_ID, DOMAIN
_LOGGER = logging.getLogger(__name__)
async def validate_input(auth: Auth) -> None:
async def validate_input(blink: Blink) -> None:
"""Validate the user input allows us to connect."""
try:
await auth.startup()
await blink.start()
except (LoginError, TokenRefreshFailed) as err:
raise InvalidAuth from err
if auth.check_key_required():
raise Require2FA
async def _send_blink_2fa_pin(hass: HomeAssistant, auth: Auth, pin: str | None) -> bool:
async def _send_blink_2fa_pin(blink: Blink, pin: str | None) -> bool:
"""Send 2FA pin to blink servers."""
blink = Blink(session=async_get_clientsession(hass))
blink.auth = auth
blink.setup_login_ids()
blink.setup_urls()
return await auth.send_auth_key(blink, pin)
await blink.send_2fa_code(pin)
return True
class BlinkConfigFlow(ConfigFlow, domain=DOMAIN):
@@ -48,6 +48,23 @@ class BlinkConfigFlow(ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize the blink flow."""
self.auth: Auth | None = None
self.blink: Blink | None = None
async def _handle_user_input(self, user_input: dict[str, Any]):
"""Handle user input."""
self.auth = Auth(
{**user_input, "device_id": DEVICE_ID},
no_prompt=True,
session=async_get_clientsession(self.hass),
)
self.blink = Blink(session=async_get_clientsession(self.hass))
self.blink.auth = self.auth
await self.async_set_unique_id(user_input[CONF_USERNAME])
if self.source not in (SOURCE_REAUTH, SOURCE_RECONFIGURE):
self._abort_if_unique_id_configured()
await validate_input(self.blink)
return self._async_finish_flow()
async def async_step_user(
self, user_input: dict[str, Any] | None = None
@@ -55,19 +72,9 @@ class BlinkConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a flow initiated by the user."""
errors = {}
if user_input is not None:
self.auth = Auth(
{**user_input, "device_id": DEVICE_ID},
no_prompt=True,
session=async_get_clientsession(self.hass),
)
await self.async_set_unique_id(user_input[CONF_USERNAME])
if self.source != SOURCE_REAUTH:
self._abort_if_unique_id_configured()
try:
await validate_input(self.auth)
return self._async_finish_flow()
except Require2FA:
return await self._handle_user_input(user_input)
except BlinkTwoFARequiredError:
return await self.async_step_2fa()
except InvalidAuth:
errors["base"] = "invalid_auth"
@@ -93,19 +100,16 @@ class BlinkConfigFlow(ConfigFlow, domain=DOMAIN):
errors = {}
if user_input is not None:
try:
valid_token = await _send_blink_2fa_pin(
self.hass, self.auth, user_input.get(CONF_PIN)
)
await _send_blink_2fa_pin(self.blink, user_input.get(CONF_PIN))
except BlinkSetupError:
errors["base"] = "cannot_connect"
except TokenRefreshFailed:
errors["base"] = "invalid_access_token"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
if valid_token:
return self._async_finish_flow()
errors["base"] = "invalid_access_token"
return self.async_show_form(
step_id="2fa",
@@ -118,19 +122,89 @@ class BlinkConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon migration of old entries."""
return await self.async_step_user(dict(entry_data))
"""Perform reauth after an authentication error."""
return await self.async_step_reauth_confirm(None)
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reauth confirmation."""
errors = {}
if user_input is not None:
try:
return await self._handle_user_input(user_input)
except BlinkTwoFARequiredError:
return await self.async_step_2fa()
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
config_entry = self._get_reauth_entry()
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema(
{
vol.Required(
CONF_USERNAME, default=config_entry.data[CONF_USERNAME]
): str,
vol.Required(
CONF_PASSWORD, default=config_entry.data[CONF_PASSWORD]
): str,
}
),
errors=errors,
description_placeholders={"username": config_entry.data[CONF_USERNAME]},
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reconfiguration initiated by the user."""
errors = {}
if user_input is not None:
try:
return await self._handle_user_input(user_input)
except BlinkTwoFARequiredError:
return await self.async_step_2fa()
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
config_entry = self._get_reconfigure_entry()
return self.async_show_form(
step_id="reconfigure",
data_schema=vol.Schema(
{
vol.Required(
CONF_USERNAME, default=config_entry.data[CONF_USERNAME]
): str,
vol.Required(
CONF_PASSWORD, default=config_entry.data[CONF_PASSWORD]
): str,
}
),
errors=errors,
)
@callback
def _async_finish_flow(self) -> ConfigFlowResult:
"""Finish with setup."""
assert self.auth
if self.source in (SOURCE_REAUTH, SOURCE_RECONFIGURE):
return self.async_update_reload_and_abort(
self._get_reauth_entry()
if self.source == SOURCE_REAUTH
else self._get_reconfigure_entry(),
data_updates=self.auth.login_attributes,
)
return self.async_create_entry(title=DOMAIN, data=self.auth.login_attributes)
class Require2FA(HomeAssistantError):
"""Error to indicate we require 2FA."""
class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""

24
homeassistant/components/blink/coordinator.py
View File

@@ -6,10 +6,17 @@ from datetime import timedelta
import logging
from typing import Any
from aiohttp import ClientError
from blinkpy.auth import BlinkTwoFARequiredError, UnauthorizedError
from blinkpy.blinkpy import Blink
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryError,
ConfigEntryNotReady,
)
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
@@ -38,6 +45,23 @@ class BlinkUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
update_interval=timedelta(seconds=SCAN_INTERVAL),
)
async def _async_setup(self):
"""Set up the coordinator."""
try:
await self.api.start()
except (ClientError, TimeoutError) as ex:
raise ConfigEntryNotReady("Can not connect to host") from ex
except (BlinkTwoFARequiredError, UnauthorizedError) as ex:
raise ConfigEntryAuthFailed("Required Blink re-authentication") from ex
except Exception as ex:
raise ConfigEntryError("Unknown error connecting to Blink") from ex
if not self.api.available:
raise ConfigEntryNotReady
async def _async_update_data(self) -> dict[str, Any]:
"""Async update wrapper."""
try:
return await self.api.refresh(force=True)
except UnauthorizedError as ex:
raise ConfigEntryAuthFailed("Blink API authorization failed") from ex

2
homeassistant/components/blink/manifest.json
View File

@@ -20,5 +20,5 @@
"documentation": "https://www.home-assistant.io/integrations/blink",
"iot_class": "cloud_polling",
"loggers": ["blinkpy"],
"requirements": ["blinkpy==0.23.0"]
"requirements": ["blinkpy==0.24.1"]
}

36
homeassistant/components/blink/services.py
View File

@@ -4,14 +4,12 @@ from __future__ import annotations
import voluptuous as vol
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_CONFIG_ENTRY_ID, CONF_PIN
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import config_validation as cv
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, issue_registry as ir
from .const import DOMAIN, SERVICE_SEND_PIN
from .coordinator import BlinkConfigEntry
SERVICE_SEND_PIN_SCHEMA = vol.Schema(
{
@@ -23,24 +21,24 @@ SERVICE_SEND_PIN_SCHEMA = vol.Schema(
async def _send_pin(call: ServiceCall) -> None:
"""Call blink to send new pin."""
config_entry: BlinkConfigEntry | None
for entry_id in call.data[ATTR_CONFIG_ENTRY_ID]:
if not (config_entry := call.hass.config_entries.async_get_entry(entry_id)):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="integration_not_found",
translation_placeholders={"target": DOMAIN},
# Create repair issue to inform user about service removal
ir.async_create_issue(
call.hass,
DOMAIN,
"service_send_pin_deprecation",
is_fixable=False,
issue_domain=DOMAIN,
severity=ir.IssueSeverity.ERROR,
breaks_in_ha_version="2026.5.0",
translation_key="service_send_pin_deprecation",
translation_placeholders={"service_name": f"{DOMAIN}.{SERVICE_SEND_PIN}"},
)
if config_entry.state != ConfigEntryState.LOADED:
# Service has been removed - raise exception
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="not_loaded",
translation_placeholders={"target": config_entry.title},
)
coordinator = config_entry.runtime_data
await coordinator.api.auth.send_auth_key(
coordinator.api,
call.data[CONF_PIN],
translation_key="service_removed",
translation_placeholders={"service_name": f"{DOMAIN}.{SERVICE_SEND_PIN}"},
)

18
homeassistant/components/blink/strings.json
View File

@@ -1,7 +1,8 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
@@ -17,6 +18,14 @@
"description": "Enter the PIN sent via email or SMS",
"title": "Two-factor authentication"
},
"reauth_confirm": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
},
"description": "The credentials for {username} need to be updated",
"title": "Re-authenticate Blink"
},
"user": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
@@ -73,6 +82,9 @@
},
"not_loaded": {
"message": "{target} is not loaded."
},
"service_removed": {
"message": "The service {service_name} has been removed and is no longer needed. Home Assistant will automatically prompt for reauthentication when required."
}
},
"issues": {
@@ -86,6 +98,10 @@
}
},
"title": "Blink update service is being removed"
},
"service_send_pin_deprecation": {
"description": "The service {service_name} has been removed and is no longer needed. When a new two-factor authentication code is required, Home Assistant will automatically prompt you to reauthenticate through the integration configuration. Please remove any automations or scripts that call this service.",
"title": "Blink send PIN service has been removed"
}
},
"options": {

10
homeassistant/components/blink/switch.py
View File

@@ -4,13 +4,15 @@ from __future__ import annotations
from typing import Any
from blinkpy.auth import UnauthorizedError
from homeassistant.components.switch import (
SwitchDeviceClass,
SwitchEntity,
SwitchEntityDescription,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
@@ -77,6 +79,9 @@ class BlinkSwitch(CoordinatorEntity[BlinkUpdateCoordinator], SwitchEntity):
translation_domain=DOMAIN,
translation_key="failed_arm_motion",
) from er
except UnauthorizedError as er:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise ConfigEntryAuthFailed("Blink authorization failed") from er
await self.coordinator.async_refresh()
@@ -90,6 +95,9 @@ class BlinkSwitch(CoordinatorEntity[BlinkUpdateCoordinator], SwitchEntity):
translation_domain=DOMAIN,
translation_key="failed_disarm_motion",
) from er
except UnauthorizedError as er:
self.coordinator.config_entry.async_start_reauth(self.hass)
raise ConfigEntryAuthFailed("Blink authorization failed") from er
await self.coordinator.async_refresh()

2
requirements_all.txt generated
View File

@@ -648,7 +648,7 @@ bleak==1.0.1
blebox-uniapi==2.5.0
# homeassistant.components.blink
blinkpy==0.23.0
blinkpy==0.24.1
# homeassistant.components.bitcoin
blockchain==1.4.4

2
requirements_test_all.txt generated
View File

@@ -582,7 +582,7 @@ bleak==1.0.1
blebox-uniapi==2.5.0
# homeassistant.components.blink
blinkpy==0.23.0
blinkpy==0.24.1
# homeassistant.components.blue_current
bluecurrent-api==1.3.1

2
tests/components/blink/conftest.py
View File

@@ -71,8 +71,6 @@ def blink_api_fixture(camera) -> MagicMock:
def blink_auth_api_fixture() -> MagicMock:
"""Set up Blink API fixture."""
mock_blink_auth_api = create_autospec(blinkpy.auth.Auth, instance=True)
mock_blink_auth_api.check_key_required.return_value = False
mock_blink_auth_api.send_auth_key = AsyncMock(return_value=True)
with patch("homeassistant.components.blink.Auth", autospec=True) as class_mock:
class_mock.return_value = mock_blink_auth_api

164
tests/components/blink/test_config_flow.py
View File

@@ -2,7 +2,7 @@
from unittest.mock import patch
from blinkpy.auth import LoginError
from blinkpy.auth import BlinkTwoFARequiredError, LoginError, TokenRefreshFailed
from blinkpy.blinkpy import BlinkSetupError
from homeassistant import config_entries
@@ -13,78 +13,6 @@ from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with (
patch("homeassistant.components.blink.config_flow.Auth.startup"),
patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
),
patch(
"homeassistant.components.blink.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"username": "blink@example.com", "password": "example"},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "blink"
assert result2["result"].unique_id == "blink@example.com"
assert result2["data"] == {
"username": "blink@example.com",
"password": "example",
"device_id": "Home Assistant",
"token": None,
"host": None,
"account_id": None,
"client_id": None,
"region_id": None,
"user_id": None,
}
assert len(mock_setup_entry.mock_calls) == 1
# Now check for duplicates
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with (
patch("homeassistant.components.blink.config_flow.Auth.startup"),
patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
),
patch(
"homeassistant.components.blink.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"username": "blink@example.com", "password": "example"},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "already_configured"
assert len(mock_setup_entry.mock_calls) == 0
async def test_form_2fa(hass: HomeAssistant) -> None:
"""Test we get the 2fa form."""
@@ -92,12 +20,9 @@ async def test_form_2fa(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with (
patch("homeassistant.components.blink.config_flow.Auth.startup"),
patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=True,
),
with patch(
"homeassistant.components.blink.config_flow.Blink.start",
side_effect=BlinkTwoFARequiredError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
@@ -108,13 +33,9 @@ async def test_form_2fa(hass: HomeAssistant) -> None:
assert result2["step_id"] == "2fa"
with (
patch("homeassistant.components.blink.config_flow.Auth.startup"),
patch("homeassistant.components.blink.config_flow.Blink.start"),
patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
),
patch(
"homeassistant.components.blink.config_flow.Auth.send_auth_key",
"homeassistant.components.blink.config_flow.Blink.send_2fa_code",
return_value=True,
),
patch(
@@ -143,12 +64,9 @@ async def test_form_2fa_connect_error(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with (
patch("homeassistant.components.blink.config_flow.Auth.startup"),
patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=True,
),
with patch(
"homeassistant.components.blink.config_flow.Blink.start",
side_effect=BlinkTwoFARequiredError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
@@ -159,17 +77,9 @@ async def test_form_2fa_connect_error(hass: HomeAssistant) -> None:
assert result2["step_id"] == "2fa"
with (
patch("homeassistant.components.blink.config_flow.Auth.startup"),
patch("homeassistant.components.blink.config_flow.Blink.start"),
patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
),
patch(
"homeassistant.components.blink.config_flow.Auth.send_auth_key",
return_value=True,
),
patch(
"homeassistant.components.blink.config_flow.Blink.setup_urls",
"homeassistant.components.blink.config_flow.Blink.send_2fa_code",
side_effect=BlinkSetupError,
),
patch(
@@ -192,12 +102,9 @@ async def test_form_2fa_invalid_key(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with (
patch("homeassistant.components.blink.config_flow.Auth.startup"),
patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=True,
),
with patch(
"homeassistant.components.blink.config_flow.Blink.start",
side_effect=BlinkTwoFARequiredError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
@@ -209,19 +116,11 @@ async def test_form_2fa_invalid_key(hass: HomeAssistant) -> None:
with (
patch(
"homeassistant.components.blink.config_flow.Auth.startup",
"homeassistant.components.blink.config_flow.Blink.start",
),
patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
),
patch(
"homeassistant.components.blink.config_flow.Auth.send_auth_key",
return_value=False,
),
patch(
"homeassistant.components.blink.config_flow.Blink.setup_urls",
return_value=True,
"homeassistant.components.blink.config_flow.Blink.send_2fa_code",
side_effect=TokenRefreshFailed,
),
patch(
"homeassistant.components.blink.async_setup_entry",
@@ -243,12 +142,9 @@ async def test_form_2fa_unknown_error(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with (
patch("homeassistant.components.blink.config_flow.Auth.startup"),
patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=True,
),
with patch(
"homeassistant.components.blink.config_flow.Blink.start",
side_effect=BlinkTwoFARequiredError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
@@ -259,18 +155,10 @@ async def test_form_2fa_unknown_error(hass: HomeAssistant) -> None:
assert result2["step_id"] == "2fa"
with (
patch("homeassistant.components.blink.config_flow.Auth.startup"),
patch("homeassistant.components.blink.config_flow.Blink.start"),
patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
),
patch(
"homeassistant.components.blink.config_flow.Auth.send_auth_key",
return_value=True,
),
patch(
"homeassistant.components.blink.config_flow.Blink.setup_urls",
side_effect=KeyError,
"homeassistant.components.blink.config_flow.Blink.send_2fa_code",
side_effect=Exception,
),
patch(
"homeassistant.components.blink.async_setup_entry",
@@ -292,7 +180,7 @@ async def test_form_invalid_auth(hass: HomeAssistant) -> None:
)
with patch(
"homeassistant.components.blink.config_flow.Auth.startup",
"homeassistant.components.blink.config_flow.Blink.start",
side_effect=LoginError,
):
result2 = await hass.config_entries.flow.async_configure(
@@ -310,7 +198,7 @@ async def test_form_unknown_error(hass: HomeAssistant) -> None:
)
with patch(
"homeassistant.components.blink.config_flow.Auth.startup",
"homeassistant.components.blink.config_flow.Blink.start",
side_effect=KeyError,
):
result2 = await hass.config_entries.flow.async_configure(
@@ -330,4 +218,4 @@ async def test_reauth_shows_user_step(hass: HomeAssistant) -> None:
mock_entry.add_to_hass(hass)
result = await mock_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["step_id"] == "reauth_confirm"

17
tests/components/blink/test_init.py
View File

@@ -1,16 +1,12 @@
"""Test the Blink init."""
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import AsyncMock, MagicMock
from aiohttp import ClientError
from blinkpy.auth import LoginError
import pytest
from homeassistant.components.blink.const import (
DOMAIN,
SERVICE_SAVE_VIDEO,
SERVICE_SEND_PIN,
)
from homeassistant.components.blink.const import DOMAIN, SERVICE_SAVE_VIDEO
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
@@ -52,14 +48,10 @@ async def test_setup_not_ready_authkey_required(
) -> None:
"""Test setup failed because 2FA is needed to connect to the Blink system."""
mock_blink_auth_api.check_key_required = MagicMock(return_value=True)
mock_blink_auth_api.send_auth_key = AsyncMock(return_value=False)
mock_blink_api.start = AsyncMock(side_effect=LoginError)
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.blink.config_flow.Auth.startup",
side_effect=LoginError,
):
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
@@ -81,7 +73,6 @@ async def test_unload_entry(
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
assert hass.services.has_service(DOMAIN, SERVICE_SAVE_VIDEO)
assert hass.services.has_service(DOMAIN, SERVICE_SEND_PIN)
async def test_migrate_V0(

84
tests/components/blink/test_services.py
View File

@@ -1,6 +1,6 @@
"""Test the Blink services."""
from unittest.mock import AsyncMock, MagicMock, Mock
from unittest.mock import MagicMock
import pytest
@@ -9,6 +9,7 @@ from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_CONFIG_ENTRY_ID, CONF_PIN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import issue_registry as ir
from tests.common import MockConfigEntry
@@ -32,15 +33,30 @@ async def test_pin_service_calls(
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_blink_api.refresh.call_count == 1
issue_registry = ir.async_get(hass)
# Service should always raise an exception and create a repair issue
with pytest.raises(
HomeAssistantError, match="The service blink.send_pin has been removed"
):
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_PIN,
{ATTR_CONFIG_ENTRY_ID: [mock_config_entry.entry_id], CONF_PIN: PIN},
blocking=True,
)
assert mock_blink_api.auth.send_auth_key.assert_awaited_once
with pytest.raises(HomeAssistantError):
# Verify repair issue was created
issues = issue_registry.issues
assert len(issues) == 1
issue = next(iter(issues.values()))
assert issue.issue_id == "service_send_pin_deprecation"
assert issue.domain == DOMAIN
# Service should still raise error with bad config ID
with pytest.raises(
HomeAssistantError, match="The service blink.send_pin has been removed"
):
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_PIN,
@@ -49,63 +65,53 @@ async def test_pin_service_calls(
)
async def test_service_pin_called_with_non_blink_device(
async def test_service_pin_creates_repair_issue(
hass: HomeAssistant,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test pin service calls with non blink device."""
"""Test that the send PIN service creates a repair issue."""
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"
other_mock_config_entry = MockConfigEntry(
title="Not Blink", domain=other_domain, entry_id=other_config_id
)
other_mock_config_entry.add_to_hass(hass)
issue_registry = ir.async_get(hass)
hass.config.is_allowed_path = Mock(return_value=True)
mock_blink_api.cameras = {CAMERA_NAME: AsyncMock()}
# Initially no issues
assert len(issue_registry.issues) == 0
parameters = {
ATTR_CONFIG_ENTRY_ID: [other_mock_config_entry.entry_id],
CONF_PIN: PIN,
}
with pytest.raises(HomeAssistantError):
# Call the service (should fail but create repair issue)
with pytest.raises(
HomeAssistantError, match="The service blink.send_pin has been removed"
):
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_PIN,
parameters,
{ATTR_CONFIG_ENTRY_ID: [mock_config_entry.entry_id], CONF_PIN: PIN},
blocking=True,
)
# Verify repair issue was created
issues = issue_registry.issues
assert len(issues) == 1
issue = next(iter(issues.values()))
assert issue.issue_id == "service_send_pin_deprecation"
assert issue.domain == DOMAIN
assert issue.severity == ir.IssueSeverity.ERROR
assert not issue.is_fixable
async def test_service_pin_called_with_unloaded_entry(
hass: HomeAssistant,
mock_blink_api: MagicMock,
mock_blink_auth_api: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test pin service calls with not ready 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()
mock_config_entry.mock_state(hass, ConfigEntryState.SETUP_ERROR)
hass.config.is_allowed_path = Mock(return_value=True)
mock_blink_api.cameras = {CAMERA_NAME: AsyncMock()}
parameters = {ATTR_CONFIG_ENTRY_ID: [mock_config_entry.entry_id], CONF_PIN: PIN}
with pytest.raises(HomeAssistantError):
# Call service again - should not create duplicate issue
with pytest.raises(
HomeAssistantError, match="The service blink.send_pin has been removed"
):
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_PIN,
parameters,
{ATTR_CONFIG_ENTRY_ID: [mock_config_entry.entry_id], CONF_PIN: PIN},
blocking=True,
)
# Still only one issue
assert len(issue_registry.issues) == 1