mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 08:47:57 +00:00
Add config flow to canary (#40055)
* Create config_flow.py * Update config_flow.py * work on config flow * Update test_config_flow.py * Update __init__.py * Update camera.py * Update test_config_flow.py * Update test_config_flow.py * Update config_flow.py * Update conftest.py * Update test_config_flow.py * Update test_init.py * Update test_init.py * Apply suggestions from code review * Update camera.py * Update test_init.py * Update camera.py * Update __init__.py * Update test_init.py * Update test_init.py * Update __init__.py * Update __init__.py * Apply suggestions from code review * Update __init__.py * Update test_init.py * Update __init__.py * Update __init__.py * Update config_flow.py * Update test_config_flow.py * Update config_flow.py * Update config_flow.py
This commit is contained in:
Chris Talkington
GitHub
parent
94dfb66824
commit
f563068ce6
@ -1,4 +1,5 @@
|
||||
"""Support for Canary devices."""
|
||||
import asyncio
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
@ -6,20 +7,26 @@ from canary.api import Api
|
||||
from requests import ConnectTimeout, HTTPError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.camera.const import DOMAIN as CAMERA_DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
|
||||
from homeassistant.helpers import discovery
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
from homeassistant.util import Throttle
|
||||
|
||||
from .const import (
|
||||
CONF_FFMPEG_ARGUMENTS,
|
||||
DATA_CANARY,
|
||||
DATA_UNDO_UPDATE_LISTENER,
|
||||
DEFAULT_FFMPEG_ARGUMENTS,
|
||||
DEFAULT_TIMEOUT,
|
||||
DOMAIN,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
NOTIFICATION_ID = "canary_notification"
|
||||
NOTIFICATION_TITLE = "Canary Setup"
|
||||
|
||||
DOMAIN = "canary"
|
||||
DATA_CANARY = "canary"
|
||||
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=30)
|
||||
DEFAULT_TIMEOUT = 10
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
@ -34,33 +41,101 @@ CONFIG_SCHEMA = vol.Schema(
|
||||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
CANARY_COMPONENTS = ["alarm_control_panel", "camera", "sensor"]
|
||||
PLATFORMS = ["alarm_control_panel", "camera", "sensor"]
|
||||
|
||||
|
||||
def setup(hass, config):
|
||||
"""Set up the Canary component."""
|
||||
conf = config[DOMAIN]
|
||||
username = conf[CONF_USERNAME]
|
||||
password = conf[CONF_PASSWORD]
|
||||
timeout = conf[CONF_TIMEOUT]
|
||||
async def async_setup(hass: HomeAssistantType, config: dict) -> bool:
|
||||
"""Set up the Canary integration."""
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
|
||||
if hass.config_entries.async_entries(DOMAIN):
|
||||
return True
|
||||
|
||||
ffmpeg_arguments = DEFAULT_FFMPEG_ARGUMENTS
|
||||
if CAMERA_DOMAIN in config:
|
||||
camera_config = next(
|
||||
(item for item in config[CAMERA_DOMAIN] if item["platform"] == DOMAIN),
|
||||
None,
|
||||
)
|
||||
|
||||
if camera_config:
|
||||
ffmpeg_arguments = camera_config.get(
|
||||
CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS
|
||||
)
|
||||
|
||||
if DOMAIN in config:
|
||||
if ffmpeg_arguments != DEFAULT_FFMPEG_ARGUMENTS:
|
||||
config[DOMAIN][CONF_FFMPEG_ARGUMENTS] = ffmpeg_arguments
|
||||
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=config[DOMAIN],
|
||||
)
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
|
||||
"""Set up Canary from a config entry."""
|
||||
if not entry.options:
|
||||
options = {
|
||||
CONF_FFMPEG_ARGUMENTS: entry.data.get(
|
||||
CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS
|
||||
),
|
||||
CONF_TIMEOUT: entry.data.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
|
||||
}
|
||||
hass.config_entries.async_update_entry(entry, options=options)
|
||||
|
||||
try:
|
||||
hass.data[DATA_CANARY] = CanaryData(username, password, timeout)
|
||||
except (ConnectTimeout, HTTPError) as ex:
|
||||
_LOGGER.error("Unable to connect to Canary service: %s", str(ex))
|
||||
hass.components.persistent_notification.create(
|
||||
f"Error: {ex}<br />You will need to restart hass after fixing.",
|
||||
title=NOTIFICATION_TITLE,
|
||||
notification_id=NOTIFICATION_ID,
|
||||
canary_data = CanaryData(
|
||||
entry.data[CONF_USERNAME],
|
||||
entry.data[CONF_PASSWORD],
|
||||
entry.options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
|
||||
)
|
||||
return False
|
||||
except (ConnectTimeout, HTTPError) as error:
|
||||
_LOGGER.error("Unable to connect to Canary service: %s", str(error))
|
||||
raise ConfigEntryNotReady from error
|
||||
|
||||
for component in CANARY_COMPONENTS:
|
||||
discovery.load_platform(hass, component, DOMAIN, {}, config)
|
||||
undo_listener = entry.add_update_listener(_async_update_listener)
|
||||
|
||||
hass.data[DOMAIN][entry.entry_id] = {
|
||||
DATA_CANARY: canary_data,
|
||||
DATA_UNDO_UPDATE_LISTENER: undo_listener,
|
||||
}
|
||||
|
||||
for component in PLATFORMS:
|
||||
hass.async_create_task(
|
||||
hass.config_entries.async_forward_entry_setup(entry, component)
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok = all(
|
||||
await asyncio.gather(
|
||||
*[
|
||||
hass.config_entries.async_forward_entry_unload(entry, component)
|
||||
for component in PLATFORMS
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN][entry.entry_id][DATA_UNDO_UPDATE_LISTENER]()
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
|
||||
return unload_ok
|
||||
|
||||
|
||||
async def _async_update_listener(hass: HomeAssistantType, entry: ConfigEntry) -> None:
|
||||
"""Handle options update."""
|
||||
await hass.config_entries.async_reload(entry.entry_id)
|
||||
|
||||
|
||||
class CanaryData:
|
||||
"""Get the latest data and update the states."""
|
||||
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
"""Support for Canary alarm."""
|
||||
import logging
|
||||
from typing import Callable, List
|
||||
|
||||
from canary.api import LOCATION_MODE_AWAY, LOCATION_MODE_HOME, LOCATION_MODE_NIGHT
|
||||
|
||||
@ -9,24 +10,32 @@ from homeassistant.components.alarm_control_panel.const import (
|
||||
SUPPORT_ALARM_ARM_HOME,
|
||||
SUPPORT_ALARM_ARM_NIGHT,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
STATE_ALARM_ARMED_AWAY,
|
||||
STATE_ALARM_ARMED_HOME,
|
||||
STATE_ALARM_ARMED_NIGHT,
|
||||
STATE_ALARM_DISARMED,
|
||||
)
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from . import DATA_CANARY
|
||||
from . import CanaryData
|
||||
from .const import DATA_CANARY, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the Canary alarms."""
|
||||
data = hass.data[DATA_CANARY]
|
||||
devices = [CanaryAlarm(data, location.location_id) for location in data.locations]
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType,
|
||||
entry: ConfigEntry,
|
||||
async_add_entities: Callable[[List[Entity], bool], None],
|
||||
) -> None:
|
||||
"""Set up Canary alarm control panels based on a config entry."""
|
||||
data: CanaryData = hass.data[DOMAIN][entry.entry_id][DATA_CANARY]
|
||||
alarms = [CanaryAlarm(data, location.location_id) for location in data.locations]
|
||||
|
||||
add_entities(devices, True)
|
||||
async_add_entities(alarms, True)
|
||||
|
||||
|
||||
class CanaryAlarm(AlarmControlPanelEntity):
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
import asyncio
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Callable, List
|
||||
|
||||
from haffmpeg.camera import CameraMjpeg
|
||||
from haffmpeg.tools import IMAGE_JPEG, ImageFrame
|
||||
@ -9,47 +10,59 @@ import voluptuous as vol
|
||||
|
||||
from homeassistant.components.camera import PLATFORM_SCHEMA, Camera
|
||||
from homeassistant.components.ffmpeg import DATA_FFMPEG
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.aiohttp_client import async_aiohttp_proxy_stream
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
from homeassistant.util import Throttle
|
||||
|
||||
from . import DATA_CANARY, DEFAULT_TIMEOUT
|
||||
from . import CanaryData
|
||||
from .const import (
|
||||
CONF_FFMPEG_ARGUMENTS,
|
||||
DATA_CANARY,
|
||||
DEFAULT_FFMPEG_ARGUMENTS,
|
||||
DEFAULT_TIMEOUT,
|
||||
DOMAIN,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONF_FFMPEG_ARGUMENTS = "ffmpeg_arguments"
|
||||
DEFAULT_ARGUMENTS = "-pred 1"
|
||||
|
||||
MIN_TIME_BETWEEN_SESSION_RENEW = timedelta(seconds=90)
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{vol.Optional(CONF_FFMPEG_ARGUMENTS, default=DEFAULT_ARGUMENTS): cv.string}
|
||||
{vol.Optional(CONF_FFMPEG_ARGUMENTS, default=DEFAULT_FFMPEG_ARGUMENTS): cv.string}
|
||||
)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the Canary sensors."""
|
||||
if discovery_info is not None:
|
||||
return
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType,
|
||||
entry: ConfigEntry,
|
||||
async_add_entities: Callable[[List[Entity], bool], None],
|
||||
) -> None:
|
||||
"""Set up Canary sensors based on a config entry."""
|
||||
data: CanaryData = hass.data[DOMAIN][entry.entry_id][DATA_CANARY]
|
||||
|
||||
data = hass.data[DATA_CANARY]
|
||||
devices = []
|
||||
ffmpeg_arguments = entry.options.get(
|
||||
CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS
|
||||
)
|
||||
cameras = []
|
||||
|
||||
for location in data.locations:
|
||||
for device in location.devices:
|
||||
if device.is_online:
|
||||
devices.append(
|
||||
cameras.append(
|
||||
CanaryCamera(
|
||||
hass,
|
||||
data,
|
||||
location,
|
||||
device,
|
||||
DEFAULT_TIMEOUT,
|
||||
config[CONF_FFMPEG_ARGUMENTS],
|
||||
ffmpeg_arguments,
|
||||
)
|
||||
)
|
||||
|
||||
add_entities(devices, True)
|
||||
async_add_entities(cameras, True)
|
||||
|
||||
|
||||
class CanaryCamera(Camera):
|
||||
|
||||
121
homeassistant/components/canary/config_flow.py
Normal file
121
homeassistant/components/canary/config_flow.py
Normal file
@ -0,0 +1,121 @@
|
||||
"""Config flow for Canary."""
|
||||
import logging
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from canary.api import Api
|
||||
from requests import ConnectTimeout, HTTPError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import CONN_CLASS_CLOUD_POLL, ConfigFlow, OptionsFlow
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
|
||||
|
||||
from .const import CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS, DEFAULT_TIMEOUT
|
||||
from .const import DOMAIN # pylint: disable=unused-import
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def validate_input(hass: HomeAssistantType, data: dict) -> Dict[str, Any]:
|
||||
"""Validate the user input allows us to connect.
|
||||
|
||||
Data has the keys from DATA_SCHEMA with values provided by the user.
|
||||
"""
|
||||
# constructor does login call
|
||||
Api(
|
||||
data[CONF_USERNAME],
|
||||
data[CONF_PASSWORD],
|
||||
data.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class CanaryConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Canary."""
|
||||
|
||||
VERSION = 1
|
||||
CONNECTION_CLASS = CONN_CLASS_CLOUD_POLL
|
||||
|
||||
@staticmethod
|
||||
@callback
|
||||
def async_get_options_flow(config_entry):
|
||||
"""Get the options flow for this handler."""
|
||||
return CanaryOptionsFlowHandler(config_entry)
|
||||
|
||||
async def async_step_import(
|
||||
self, user_input: Optional[ConfigType] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Handle a flow initiated by configuration file."""
|
||||
return await self.async_step_user(user_input)
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: Optional[ConfigType] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Handle a flow initiated by the user."""
|
||||
if self._async_current_entries():
|
||||
return self.async_abort(reason="single_instance_allowed")
|
||||
|
||||
errors = {}
|
||||
default_username = ""
|
||||
|
||||
if user_input is not None:
|
||||
if CONF_TIMEOUT not in user_input:
|
||||
user_input[CONF_TIMEOUT] = DEFAULT_TIMEOUT
|
||||
|
||||
default_username = user_input[CONF_USERNAME]
|
||||
|
||||
try:
|
||||
await self.hass.async_add_executor_job(
|
||||
validate_input, self.hass, user_input
|
||||
)
|
||||
except (ConnectTimeout, HTTPError):
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
return self.async_abort(reason="unknown")
|
||||
else:
|
||||
return self.async_create_entry(
|
||||
title=user_input[CONF_USERNAME],
|
||||
data=user_input,
|
||||
)
|
||||
|
||||
data_schema = {
|
||||
vol.Required(CONF_USERNAME, default=default_username): str,
|
||||
vol.Required(CONF_PASSWORD): str,
|
||||
}
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=vol.Schema(data_schema),
|
||||
errors=errors or {},
|
||||
)
|
||||
|
||||
|
||||
class CanaryOptionsFlowHandler(OptionsFlow):
|
||||
"""Handle Canary client options."""
|
||||
|
||||
def __init__(self, config_entry):
|
||||
"""Initialize options flow."""
|
||||
self.config_entry = config_entry
|
||||
|
||||
async def async_step_init(self, user_input: Optional[ConfigType] = None):
|
||||
"""Manage Canary options."""
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(title="", data=user_input)
|
||||
|
||||
options = {
|
||||
vol.Optional(
|
||||
CONF_FFMPEG_ARGUMENTS,
|
||||
default=self.config_entry.options.get(
|
||||
CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS
|
||||
),
|
||||
): str,
|
||||
vol.Optional(
|
||||
CONF_TIMEOUT,
|
||||
default=self.config_entry.options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
|
||||
): int,
|
||||
}
|
||||
|
||||
return self.async_show_form(step_id="init", data_schema=vol.Schema(options))
|
||||
14
homeassistant/components/canary/const.py
Normal file
14
homeassistant/components/canary/const.py
Normal file
@ -0,0 +1,14 @@
|
||||
"""Constants for the Canary integration."""
|
||||
|
||||
DOMAIN = "canary"
|
||||
|
||||
# Configuration
|
||||
CONF_FFMPEG_ARGUMENTS = "ffmpeg_arguments"
|
||||
|
||||
# Data
|
||||
DATA_CANARY = "canary"
|
||||
DATA_UNDO_UPDATE_LISTENER = "undo_update_listener"
|
||||
|
||||
# Defaults
|
||||
DEFAULT_FFMPEG_ARGUMENTS = "-pred 1"
|
||||
DEFAULT_TIMEOUT = 10
|
||||
@ -4,5 +4,6 @@
|
||||
"documentation": "https://www.home-assistant.io/integrations/canary",
|
||||
"requirements": ["py-canary==0.5.0"],
|
||||
"dependencies": ["ffmpeg"],
|
||||
"codeowners": []
|
||||
"codeowners": [],
|
||||
"config_flow": true
|
||||
}
|
||||
|
||||
@ -1,6 +1,9 @@
|
||||
"""Support for Canary sensors."""
|
||||
from typing import Callable, List
|
||||
|
||||
from canary.api import SensorType
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
DEVICE_CLASS_BATTERY,
|
||||
DEVICE_CLASS_HUMIDITY,
|
||||
@ -10,8 +13,10 @@ from homeassistant.const import (
|
||||
TEMP_CELSIUS,
|
||||
)
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from . import DATA_CANARY
|
||||
from . import CanaryData
|
||||
from .const import DATA_CANARY, DOMAIN
|
||||
|
||||
SENSOR_VALUE_PRECISION = 2
|
||||
ATTR_AIR_QUALITY = "air_quality"
|
||||
@ -38,10 +43,14 @@ STATE_AIR_QUALITY_ABNORMAL = "abnormal"
|
||||
STATE_AIR_QUALITY_VERY_ABNORMAL = "very_abnormal"
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the Canary sensors."""
|
||||
data = hass.data[DATA_CANARY]
|
||||
devices = []
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType,
|
||||
entry: ConfigEntry,
|
||||
async_add_entities: Callable[[List[Entity], bool], None],
|
||||
) -> None:
|
||||
"""Set up Canary sensors based on a config entry."""
|
||||
data: CanaryData = hass.data[DOMAIN][entry.entry_id][DATA_CANARY]
|
||||
sensors = []
|
||||
|
||||
for location in data.locations:
|
||||
for device in location.devices:
|
||||
@ -49,11 +58,11 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
device_type = device.device_type
|
||||
for sensor_type in SENSOR_TYPES:
|
||||
if device_type.get("name") in sensor_type[4]:
|
||||
devices.append(
|
||||
sensors.append(
|
||||
CanarySensor(data, sensor_type, location, device)
|
||||
)
|
||||
|
||||
add_entities(devices, True)
|
||||
async_add_entities(sensors, True)
|
||||
|
||||
|
||||
class CanarySensor(Entity):
|
||||
|
||||
31
homeassistant/components/canary/strings.json
Normal file
31
homeassistant/components/canary/strings.json
Normal file
@ -0,0 +1,31 @@
|
||||
{
|
||||
"config": {
|
||||
"flow_title": "Canary: {name}",
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Connect to Canary",
|
||||
"data": {
|
||||
"username": "[%key:common::config_flow::data::username%]",
|
||||
"password": "[%key:common::config_flow::data::password%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
|
||||
},
|
||||
"abort": {
|
||||
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"data": {
|
||||
"ffmpeg_arguments": "Arguments passed to ffmpeg for cameras",
|
||||
"timeout": "Request Timeout (seconds)"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -31,6 +31,7 @@ FLOWS = [
|
||||
"broadlink",
|
||||
"brother",
|
||||
"bsblan",
|
||||
"canary",
|
||||
"cast",
|
||||
"cert_expiry",
|
||||
"control4",
|
||||
|
||||
@ -1,8 +1,73 @@
|
||||
"""Tests for the canary component."""
|
||||
"""Tests for the Canary integration."""
|
||||
from unittest.mock import MagicMock, PropertyMock
|
||||
|
||||
from canary.api import SensorType
|
||||
|
||||
from homeassistant.components.canary.const import (
|
||||
CONF_FFMPEG_ARGUMENTS,
|
||||
DEFAULT_FFMPEG_ARGUMENTS,
|
||||
DEFAULT_TIMEOUT,
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from tests.async_mock import patch
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
ENTRY_CONFIG = {
|
||||
CONF_PASSWORD: "test-password",
|
||||
CONF_USERNAME: "test-username",
|
||||
}
|
||||
|
||||
ENTRY_OPTIONS = {
|
||||
CONF_FFMPEG_ARGUMENTS: DEFAULT_FFMPEG_ARGUMENTS,
|
||||
CONF_TIMEOUT: DEFAULT_TIMEOUT,
|
||||
}
|
||||
|
||||
USER_INPUT = {
|
||||
CONF_PASSWORD: "test-password",
|
||||
CONF_USERNAME: "test-username",
|
||||
}
|
||||
|
||||
YAML_CONFIG = {
|
||||
CONF_PASSWORD: "test-password",
|
||||
CONF_USERNAME: "test-username",
|
||||
CONF_TIMEOUT: 5,
|
||||
}
|
||||
|
||||
|
||||
def _patch_async_setup(return_value=True):
|
||||
return patch(
|
||||
"homeassistant.components.canary.async_setup",
|
||||
return_value=return_value,
|
||||
)
|
||||
|
||||
|
||||
def _patch_async_setup_entry(return_value=True):
|
||||
return patch(
|
||||
"homeassistant.components.canary.async_setup_entry",
|
||||
return_value=return_value,
|
||||
)
|
||||
|
||||
|
||||
async def init_integration(
|
||||
hass: HomeAssistantType,
|
||||
*,
|
||||
data: dict = ENTRY_CONFIG,
|
||||
options: dict = ENTRY_OPTIONS,
|
||||
skip_entry_setup: bool = False,
|
||||
) -> MockConfigEntry:
|
||||
"""Set up the Canary integration in Home Assistant."""
|
||||
entry = MockConfigEntry(domain=DOMAIN, data=data, options=options)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
if not skip_entry_setup:
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
return entry
|
||||
|
||||
|
||||
def mock_device(device_id, name, is_online=True, device_type_name=None):
|
||||
"""Mock Canary Device class."""
|
||||
@ -13,6 +78,7 @@ def mock_device(device_id, name, is_online=True, device_type_name=None):
|
||||
type(device).device_type = PropertyMock(
|
||||
return_value={"id": 1, "name": device_type_name}
|
||||
)
|
||||
|
||||
return device
|
||||
|
||||
|
||||
@ -27,6 +93,7 @@ def mock_location(
|
||||
type(location).is_private = PropertyMock(return_value=is_private)
|
||||
type(location).devices = PropertyMock(return_value=devices or [])
|
||||
type(location).mode = PropertyMock(return_value=mode)
|
||||
|
||||
return location
|
||||
|
||||
|
||||
@ -36,6 +103,7 @@ def mock_mode(mode_id, name):
|
||||
type(mode).mode_id = PropertyMock(return_value=mode_id)
|
||||
type(mode).name = PropertyMock(return_value=name)
|
||||
type(mode).resource_url = PropertyMock(return_value=f"/v1/modes/{mode_id}")
|
||||
|
||||
return mode
|
||||
|
||||
|
||||
@ -44,4 +112,5 @@ def mock_reading(sensor_type, sensor_value):
|
||||
reading = MagicMock()
|
||||
type(reading).sensor_type = SensorType(sensor_type)
|
||||
type(reading).value = PropertyMock(return_value=sensor_value)
|
||||
|
||||
return reading
|
||||
|
||||
@ -32,3 +32,27 @@ def canary(hass):
|
||||
instance.set_location_mode = MagicMock(return_value=None)
|
||||
|
||||
yield mock_canary
|
||||
|
||||
|
||||
@fixture
|
||||
def canary_config_flow(hass):
|
||||
"""Mock the CanaryApi for easier config flow testing."""
|
||||
with patch.object(Api, "login", return_value=True), patch(
|
||||
"homeassistant.components.canary.config_flow.Api"
|
||||
) as mock_canary:
|
||||
instance = mock_canary.return_value = Api(
|
||||
"test-username",
|
||||
"test-password",
|
||||
1,
|
||||
)
|
||||
|
||||
instance.login = MagicMock(return_value=True)
|
||||
instance.get_entries = MagicMock(return_value=[])
|
||||
instance.get_locations = MagicMock(return_value=[])
|
||||
instance.get_location = MagicMock(return_value=None)
|
||||
instance.get_modes = MagicMock(return_value=[])
|
||||
instance.get_readings = MagicMock(return_value=[])
|
||||
instance.get_latest_readings = MagicMock(return_value=[])
|
||||
instance.set_location_mode = MagicMock(return_value=None)
|
||||
|
||||
yield mock_canary
|
||||
|
||||
@ -42,9 +42,7 @@ async def test_alarm_control_panel(hass, canary) -> None:
|
||||
instance.get_locations.return_value = [mocked_location]
|
||||
|
||||
config = {DOMAIN: {"username": "test-username", "password": "test-password"}}
|
||||
with patch(
|
||||
"homeassistant.components.canary.CANARY_COMPONENTS", ["alarm_control_panel"]
|
||||
):
|
||||
with patch("homeassistant.components.canary.PLATFORMS", ["alarm_control_panel"]):
|
||||
assert await async_setup_component(hass, DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
@ -126,9 +124,7 @@ async def test_alarm_control_panel_services(hass, canary) -> None:
|
||||
instance.get_locations.return_value = [mocked_location]
|
||||
|
||||
config = {DOMAIN: {"username": "test-username", "password": "test-password"}}
|
||||
with patch(
|
||||
"homeassistant.components.canary.CANARY_COMPONENTS", ["alarm_control_panel"]
|
||||
):
|
||||
with patch("homeassistant.components.canary.PLATFORMS", ["alarm_control_panel"]):
|
||||
assert await async_setup_component(hass, DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
121
tests/components/canary/test_config_flow.py
Normal file
121
tests/components/canary/test_config_flow.py
Normal file
@ -0,0 +1,121 @@
|
||||
"""Test the Canary config flow."""
|
||||
from requests import ConnectTimeout, HTTPError
|
||||
|
||||
from homeassistant.components.canary.const import (
|
||||
CONF_FFMPEG_ARGUMENTS,
|
||||
DEFAULT_FFMPEG_ARGUMENTS,
|
||||
DEFAULT_TIMEOUT,
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_USER
|
||||
from homeassistant.const import CONF_TIMEOUT
|
||||
from homeassistant.data_entry_flow import (
|
||||
RESULT_TYPE_ABORT,
|
||||
RESULT_TYPE_CREATE_ENTRY,
|
||||
RESULT_TYPE_FORM,
|
||||
)
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import USER_INPUT, _patch_async_setup, _patch_async_setup_entry, init_integration
|
||||
|
||||
|
||||
async def test_user_form(hass, canary_config_flow):
|
||||
"""Test we get the user initiated form."""
|
||||
await async_setup_component(hass, "persistent_notification", {})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
with _patch_async_setup() as mock_setup, _patch_async_setup_entry() as mock_setup_entry:
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
USER_INPUT,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["title"] == "test-username"
|
||||
assert result["data"] == {**USER_INPUT, CONF_TIMEOUT: DEFAULT_TIMEOUT}
|
||||
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_user_form_cannot_connect(hass, canary_config_flow):
|
||||
"""Test we handle errors that should trigger the cannot connect error."""
|
||||
canary_config_flow.side_effect = HTTPError()
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
USER_INPUT,
|
||||
)
|
||||
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
canary_config_flow.side_effect = ConnectTimeout()
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
USER_INPUT,
|
||||
)
|
||||
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
async def test_user_form_unexpected_exception(hass, canary_config_flow):
|
||||
"""Test we handle unexpected exception."""
|
||||
canary_config_flow.side_effect = Exception()
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
USER_INPUT,
|
||||
)
|
||||
|
||||
assert result["type"] == RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "unknown"
|
||||
|
||||
|
||||
async def test_user_form_single_instance_allowed(hass, canary_config_flow):
|
||||
"""Test that configuring more than one instance is rejected."""
|
||||
await init_integration(hass, skip_entry_setup=True)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
data=USER_INPUT,
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "single_instance_allowed"
|
||||
|
||||
|
||||
async def test_options_flow(hass):
|
||||
"""Test updating options."""
|
||||
entry = await init_integration(hass, skip_entry_setup=True)
|
||||
assert entry.options[CONF_FFMPEG_ARGUMENTS] == DEFAULT_FFMPEG_ARGUMENTS
|
||||
assert entry.options[CONF_TIMEOUT] == DEFAULT_TIMEOUT
|
||||
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_FFMPEG_ARGUMENTS: "-v", CONF_TIMEOUT: 7},
|
||||
)
|
||||
|
||||
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"][CONF_FFMPEG_ARGUMENTS] == "-v"
|
||||
assert result["data"][CONF_TIMEOUT] == 7
|
||||
@ -1,37 +1,82 @@
|
||||
"""The tests for the Canary component."""
|
||||
from requests import HTTPError
|
||||
from requests import ConnectTimeout
|
||||
|
||||
from homeassistant.components.canary import DOMAIN
|
||||
from homeassistant.components.camera.const import DOMAIN as CAMERA_DOMAIN
|
||||
from homeassistant.components.canary.const import CONF_FFMPEG_ARGUMENTS, DOMAIN
|
||||
from homeassistant.config_entries import (
|
||||
ENTRY_STATE_LOADED,
|
||||
ENTRY_STATE_NOT_LOADED,
|
||||
ENTRY_STATE_SETUP_RETRY,
|
||||
)
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import YAML_CONFIG, init_integration
|
||||
|
||||
from tests.async_mock import patch
|
||||
|
||||
|
||||
async def test_setup_with_valid_config(hass, canary) -> None:
|
||||
"""Test setup with valid YAML."""
|
||||
await async_setup_component(hass, "persistent_notification", {})
|
||||
config = {DOMAIN: {"username": "test-username", "password": "test-password"}}
|
||||
|
||||
async def test_import_from_yaml(hass, canary) -> None:
|
||||
"""Test import from YAML."""
|
||||
with patch(
|
||||
"homeassistant.components.canary.alarm_control_panel.setup_platform",
|
||||
return_value=True,
|
||||
), patch(
|
||||
"homeassistant.components.canary.camera.setup_platform",
|
||||
return_value=True,
|
||||
), patch(
|
||||
"homeassistant.components.canary.sensor.setup_platform",
|
||||
"homeassistant.components.canary.async_setup_entry",
|
||||
return_value=True,
|
||||
):
|
||||
assert await async_setup_component(hass, DOMAIN, config)
|
||||
assert await async_setup_component(hass, DOMAIN, {DOMAIN: YAML_CONFIG})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 1
|
||||
|
||||
async def test_setup_with_http_error(hass, canary) -> None:
|
||||
"""Test setup with HTTP error."""
|
||||
await async_setup_component(hass, "persistent_notification", {})
|
||||
config = {DOMAIN: {"username": "test-username", "password": "test-password"}}
|
||||
assert entries[0].data[CONF_USERNAME] == "test-username"
|
||||
assert entries[0].data[CONF_PASSWORD] == "test-password"
|
||||
assert entries[0].data[CONF_TIMEOUT] == 5
|
||||
|
||||
canary.side_effect = HTTPError()
|
||||
|
||||
assert not await async_setup_component(hass, DOMAIN, config)
|
||||
async def test_import_from_yaml_ffmpeg(hass, canary) -> None:
|
||||
"""Test import from YAML with ffmpeg arguments."""
|
||||
with patch(
|
||||
"homeassistant.components.canary.async_setup_entry",
|
||||
return_value=True,
|
||||
):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
DOMAIN,
|
||||
{
|
||||
DOMAIN: YAML_CONFIG,
|
||||
CAMERA_DOMAIN: [{"platform": DOMAIN, CONF_FFMPEG_ARGUMENTS: "-v"}],
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 1
|
||||
|
||||
assert entries[0].data[CONF_USERNAME] == "test-username"
|
||||
assert entries[0].data[CONF_PASSWORD] == "test-password"
|
||||
assert entries[0].data[CONF_TIMEOUT] == 5
|
||||
assert entries[0].data.get(CONF_FFMPEG_ARGUMENTS) == "-v"
|
||||
|
||||
|
||||
async def test_unload_entry(hass, canary):
|
||||
"""Test successful unload of entry."""
|
||||
entry = await init_integration(hass)
|
||||
|
||||
assert entry
|
||||
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
|
||||
assert entry.state == ENTRY_STATE_LOADED
|
||||
|
||||
assert await hass.config_entries.async_unload(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert entry.state == ENTRY_STATE_NOT_LOADED
|
||||
assert not hass.data.get(DOMAIN)
|
||||
|
||||
|
||||
async def test_async_setup_raises_entry_not_ready(hass, canary):
|
||||
"""Test that it throws ConfigEntryNotReady when exception occurs during setup."""
|
||||
canary.side_effect = ConnectTimeout()
|
||||
|
||||
entry = await init_integration(hass)
|
||||
assert entry
|
||||
assert entry.state == ENTRY_STATE_SETUP_RETRY
|
||||
|
||||
@ -42,7 +42,7 @@ async def test_sensors_pro(hass, canary) -> None:
|
||||
]
|
||||
|
||||
config = {DOMAIN: {"username": "test-username", "password": "test-password"}}
|
||||
with patch("homeassistant.components.canary.CANARY_COMPONENTS", ["sensor"]):
|
||||
with patch("homeassistant.components.canary.PLATFORMS", ["sensor"]):
|
||||
assert await async_setup_component(hass, DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
@ -101,7 +101,7 @@ async def test_sensors_attributes_pro(hass, canary) -> None:
|
||||
]
|
||||
|
||||
config = {DOMAIN: {"username": "test-username", "password": "test-password"}}
|
||||
with patch("homeassistant.components.canary.CANARY_COMPONENTS", ["sensor"]):
|
||||
with patch("homeassistant.components.canary.PLATFORMS", ["sensor"]):
|
||||
assert await async_setup_component(hass, DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
@ -155,7 +155,7 @@ async def test_sensors_flex(hass, canary) -> None:
|
||||
]
|
||||
|
||||
config = {DOMAIN: {"username": "test-username", "password": "test-password"}}
|
||||
with patch("homeassistant.components.canary.CANARY_COMPONENTS", ["sensor"]):
|
||||
with patch("homeassistant.components.canary.PLATFORMS", ["sensor"]):
|
||||
assert await async_setup_component(hass, DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user