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

Add hyperion config options flow (#43673)

Browse source
This commit is contained in:
Dermot Duffy 2020-11-30 09:38:52 -08:00 • committed by GitHub
parent 14d1466400
commit 7ad2a6be30
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 2213 additions and 156 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

173
homeassistant/components/hyperion/__init__.py
View file

@ -1 +1,174 @@
"""The Hyperion component."""
import asyncio
import logging
from typing import Any, Optional
from hyperion import client, const as hyperion_const
from pkg_resources import parse_version
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from .const import (
CONF_ON_UNLOAD,
CONF_ROOT_CLIENT,
DOMAIN,
HYPERION_RELEASES_URL,
HYPERION_VERSION_WARN_CUTOFF,
SIGNAL_INSTANCES_UPDATED,
)
PLATFORMS = [LIGHT_DOMAIN]
_LOGGER = logging.getLogger(__name__)
# Unique ID
# =========
# A config entry represents a connection to a single Hyperion server. The config entry
# unique_id is the server id returned from the Hyperion instance (a unique ID per
# server).
#
# Each server connection may create multiple entities. The unique_id for each entity is
# <server id>_<instance #>_<name>, where <server_id> will be the unique_id on the
# relevant config entry (as above), <instance #> will be the server instance # and
# <name> will be a unique identifying type name for each entity associated with this
# server/instance (e.g. "hyperion_light").
#
# The get_hyperion_unique_id method will create a per-entity unique id when given the
# server id, an instance number and a name.
# hass.data format
# ================
#
# hass.data[DOMAIN] = {
# <config_entry.entry_id>: {
# "ROOT_CLIENT": <Hyperion Client>,
# "ON_UNLOAD": [<callable>, ...],
# }
# }
def get_hyperion_unique_id(server_id: str, instance: int, name: str) -> str:
"""Get a unique_id for a Hyperion instance."""
return f"{server_id}_{instance}_{name}"
def create_hyperion_client(
*args: Any,
**kwargs: Any,
) -> client.HyperionClient:
"""Create a Hyperion Client."""
return client.HyperionClient(*args, **kwargs)
async def async_create_connect_hyperion_client(
*args: Any,
**kwargs: Any,
) -> Optional[client.HyperionClient]:
"""Create and connect a Hyperion Client."""
hyperion_client = create_hyperion_client(*args, **kwargs)
if not await hyperion_client.async_client_connect():
return None
return hyperion_client
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Hyperion component."""
hass.data[DOMAIN] = {}
return True
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Set up Hyperion from a config entry."""
host = config_entry.data[CONF_HOST]
port = config_entry.data[CONF_PORT]
token = config_entry.data.get(CONF_TOKEN)
hyperion_client = await async_create_connect_hyperion_client(
host, port, token=token
)
if not hyperion_client:
raise ConfigEntryNotReady
version = await hyperion_client.async_sysinfo_version()
if version is not None:
try:
if parse_version(version) < parse_version(HYPERION_VERSION_WARN_CUTOFF):
_LOGGER.warning(
"Using a Hyperion server version < %s is not recommended -- "
"some features may be unavailable or may not function correctly. "
"Please consider upgrading: %s",
HYPERION_VERSION_WARN_CUTOFF,
HYPERION_RELEASES_URL,
)
except ValueError:
pass
hyperion_client.set_callbacks(
{
f"{hyperion_const.KEY_INSTANCE}-{hyperion_const.KEY_UPDATE}": lambda json: (
async_dispatcher_send(
hass,
SIGNAL_INSTANCES_UPDATED.format(config_entry.entry_id),
json,
)
)
}
)
hass.data[DOMAIN][config_entry.entry_id] = {
CONF_ROOT_CLIENT: hyperion_client,
CONF_ON_UNLOAD: [],
}
# Must only listen for option updates after the setup is complete, as otherwise
# the YAML->ConfigEntry migration code triggers an options update, which causes a
# reload -- which clashes with the initial load (causing entity_id / unique_id
# clashes).
async def setup_then_listen() -> None:
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_setup(config_entry, component)
for component in PLATFORMS
]
)
hass.data[DOMAIN][config_entry.entry_id][CONF_ON_UNLOAD].append(
config_entry.add_update_listener(_async_options_updated)
)
hass.async_create_task(setup_then_listen())
return True
async def _async_options_updated(
hass: HomeAssistantType, config_entry: ConfigEntry
) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(config_entry.entry_id)
async def async_unload_entry(
hass: HomeAssistantType, config_entry: ConfigEntry
) -> bool:
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(config_entry, component)
for component in PLATFORMS
]
)
)
if unload_ok and config_entry.entry_id in hass.data[DOMAIN]:
config_data = hass.data[DOMAIN].pop(config_entry.entry_id)
for func in config_data[CONF_ON_UNLOAD]:
func()
root_client = config_data[CONF_ROOT_CLIENT]
await root_client.async_client_disconnect()
return unload_ok

445
homeassistant/components/hyperion/config_flow.py Normal file
View file

@ -0,0 +1,445 @@
"""Hyperion config flow."""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Optional
from urllib.parse import urlparse
from hyperion import client, const
import voluptuous as vol
from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_SERIAL
from homeassistant.config_entries import (
CONN_CLASS_LOCAL_PUSH,
ConfigEntry,
ConfigFlow,
OptionsFlow,
)
from homeassistant.const import CONF_BASE, CONF_HOST, CONF_ID, CONF_PORT, CONF_TOKEN
from homeassistant.core import callback
from homeassistant.helpers.typing import ConfigType
from . import create_hyperion_client
# pylint: disable=unused-import
from .const import (
CONF_AUTH_ID,
CONF_CREATE_TOKEN,
CONF_PRIORITY,
DEFAULT_ORIGIN,
DEFAULT_PRIORITY,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
_LOGGER.setLevel(logging.DEBUG)
# +------------------+ +------------------+ +--------------------+
# |Step: SSDP | |Step: user | |Step: import |
# | | | | | |
# |Input: <discovery>| |Input: <host/port>| |Input: <import data>|
# +------------------+ +------------------+ +--------------------+
# v v v
# +----------------------+-----------------------+
# Auth not | Auth |
# required? | required? |
# | v
# | +------------+
# | |Step: auth |
# | | |
# | |Input: token|
# | +------------+
# | Static |
# v token |
# <------------------+
# | |
# | | New token
# | v
# | +------------------+
# | |Step: create_token|
# | +------------------+
# | |
# | v
# | +---------------------------+ +--------------------------------+
# | |Step: create_token_external|-->|Step: create_token_external_fail|
# | +---------------------------+ +--------------------------------+
# | |
# | v
# | +-----------------------------------+
# | |Step: create_token_external_success|
# | +-----------------------------------+
# | |
# v<------------------+
# |
# v
# +-------------+ Confirm not required?
# |Step: Confirm|---------------------->+
# +-------------+ |
# | |
# v SSDP: Explicit confirm |
# +------------------------------>+
# |
# v
# +----------------+
# | Create! |
# +----------------+
# A note on choice of discovery mechanisms: Hyperion supports both Zeroconf and SSDP out
# of the box. This config flow needs two port numbers from the Hyperion instance, the
# JSON port (for the API) and the UI port (for the user to approve dynamically created
# auth tokens). With Zeroconf the port numbers for both are in different Zeroconf
# entries, and as Home Assistant only passes a single entry into the config flow, we can
# only conveniently 'see' one port or the other (which means we need to guess one port
# number). With SSDP, we get the combined block including both port numbers, so SSDP is
# the favored discovery implementation.
class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a Hyperion config flow."""
VERSION = 1
CONNECTION_CLASS = CONN_CLASS_LOCAL_PUSH
def __init__(self) -> None:
"""Instantiate config flow."""
self._data: Dict[str, Any] = {}
self._request_token_task: Optional[asyncio.Task] = None
self._auth_id: Optional[str] = None
self._require_confirm: bool = False
self._port_ui: int = const.DEFAULT_PORT_UI
def _create_client(self, raw_connection: bool = False) -> client.HyperionClient:
"""Create and connect a client instance."""
return create_hyperion_client(
self._data[CONF_HOST],
self._data[CONF_PORT],
token=self._data.get(CONF_TOKEN),
raw_connection=raw_connection,
)
async def _advance_to_auth_step_if_necessary(
self, hyperion_client: client.HyperionClient
) -> Dict[str, Any]:
"""Determine if auth is required."""
auth_resp = await hyperion_client.async_is_auth_required()
# Could not determine if auth is required.
if not auth_resp or not client.ResponseOK(auth_resp):
return self.async_abort(reason="auth_required_error")
auth_required = auth_resp.get(const.KEY_INFO, {}).get(const.KEY_REQUIRED, False)
if auth_required:
return await self.async_step_auth()
return await self.async_step_confirm()
async def async_step_import(self, import_data: ConfigType) -> Dict[str, Any]:
"""Handle a flow initiated by a YAML config import."""
self._data.update(import_data)
async with self._create_client(raw_connection=True) as hyperion_client:
if not hyperion_client:
return self.async_abort(reason="cannot_connect")
return await self._advance_to_auth_step_if_necessary(hyperion_client)
async def async_step_ssdp( # type: ignore[override]
self, discovery_info: Dict[str, Any]
) -> Dict[str, Any]:
"""Handle a flow initiated by SSDP."""
# Sample data provided by SSDP: {
# 'ssdp_location': 'http://192.168.0.1:8090/description.xml',
# 'ssdp_st': 'upnp:rootdevice',
# 'deviceType': 'urn:schemas-upnp-org:device:Basic:1',
# 'friendlyName': 'Hyperion (192.168.0.1)',
# 'manufacturer': 'Hyperion Open Source Ambient Lighting',
# 'manufacturerURL': 'https://www.hyperion-project.org',
# 'modelDescription': 'Hyperion Open Source Ambient Light',
# 'modelName': 'Hyperion',
# 'modelNumber': '2.0.0-alpha.8',
# 'modelURL': 'https://www.hyperion-project.org',
# 'serialNumber': 'f9aab089-f85a-55cf-b7c1-222a72faebe9',
# 'UDN': 'uuid:f9aab089-f85a-55cf-b7c1-222a72faebe9',
# 'ports': {
# 'jsonServer': '19444',
# 'sslServer': '8092',
# 'protoBuffer': '19445',
# 'flatBuffer': '19400'
# },
# 'presentationURL': 'index.html',
# 'iconList': {
# 'icon': {
# 'mimetype': 'image/png',
# 'height': '100',
# 'width': '100',
# 'depth': '32',
# 'url': 'img/hyperion/ssdp_icon.png'
# }
# },
# 'ssdp_usn': 'uuid:f9aab089-f85a-55cf-b7c1-222a72faebe9',
# 'ssdp_ext': '',
# 'ssdp_server': 'Raspbian GNU/Linux 10 (buster)/10 UPnP/1.0 Hyperion/2.0.0-alpha.8'}
# SSDP requires user confirmation.
self._require_confirm = True
self._data[CONF_HOST] = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
try:
self._port_ui = urlparse(discovery_info[ATTR_SSDP_LOCATION]).port
except ValueError:
self._port_ui = const.DEFAULT_PORT_UI
try:
self._data[CONF_PORT] = int(
discovery_info.get("ports", {}).get(
"jsonServer", const.DEFAULT_PORT_JSON
)
)
except ValueError:
self._data[CONF_PORT] = const.DEFAULT_PORT_JSON
hyperion_id = discovery_info.get(ATTR_UPNP_SERIAL)
if not hyperion_id:
return self.async_abort(reason="no_id")
# For discovery mechanisms, we set the unique_id as early as possible to
# avoid discovery popping up a duplicate on the screen. The unique_id is set
# authoritatively later in the flow by asking the server to confirm its id
# (which should theoretically be the same as specified here)
await self.async_set_unique_id(hyperion_id)
self._abort_if_unique_id_configured()
async with self._create_client(raw_connection=True) as hyperion_client:
if not hyperion_client:
return self.async_abort(reason="cannot_connect")
return await self._advance_to_auth_step_if_necessary(hyperion_client)
# pylint: disable=arguments-differ
async def async_step_user(
self,
user_input: Optional[ConfigType] = None,
) -> Dict[str, Any]:
"""Handle a flow initiated by the user."""
errors = {}
if user_input:
self._data.update(user_input)
async with self._create_client(raw_connection=True) as hyperion_client:
if hyperion_client:
return await self._advance_to_auth_step_if_necessary(
hyperion_client
)
errors[CONF_BASE] = "cannot_connect"
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Optional(CONF_PORT, default=const.DEFAULT_PORT_JSON): int,
}
),
errors=errors,
)
async def _cancel_request_token_task(self) -> None:
"""Cancel the request token task if it exists."""
if self._request_token_task is not None:
if not self._request_token_task.done():
self._request_token_task.cancel()
try:
await self._request_token_task
except asyncio.CancelledError:
pass
self._request_token_task = None
async def _request_token_task_func(self, auth_id: str) -> None:
"""Send an async_request_token request."""
auth_resp: Optional[Dict[str, Any]] = None
async with self._create_client(raw_connection=True) as hyperion_client:
if hyperion_client:
# The Hyperion-py client has a default timeout of 3 minutes on this request.
auth_resp = await hyperion_client.async_request_token(
comment=DEFAULT_ORIGIN, id=auth_id
)
assert self.hass
await self.hass.config_entries.flow.async_configure(
flow_id=self.flow_id, user_input=auth_resp
)
def _get_hyperion_url(self) -> str:
"""Return the URL of the Hyperion UI."""
# If this flow was kicked off by SSDP, this will be the correct frontend URL. If
# this is a manual flow instantiation, then it will be a best guess (as this
# flow does not have that information available to it). This is only used for
# approving new dynamically created tokens, so the complexity of asking the user
# manually for this information is likely not worth it (when it would only be
# used to open a URL, that the user already knows the address of).
return f"http://{self._data[CONF_HOST]}:{self._port_ui}"
async def _can_login(self) -> Optional[bool]:
"""Verify login details."""
async with self._create_client(raw_connection=True) as hyperion_client:
if not hyperion_client:
return None
return bool(
client.LoginResponseOK(
await hyperion_client.async_login(token=self._data[CONF_TOKEN])
)
)
async def async_step_auth(
self,
user_input: Optional[ConfigType] = None,
) -> Dict[str, Any]:
"""Handle the auth step of a flow."""
errors = {}
if user_input:
if user_input.get(CONF_CREATE_TOKEN):
return await self.async_step_create_token()
# Using a static token.
self._data[CONF_TOKEN] = user_input.get(CONF_TOKEN)
login_ok = await self._can_login()
if login_ok is None:
return self.async_abort(reason="cannot_connect")
if login_ok:
return await self.async_step_confirm()
errors[CONF_BASE] = "invalid_access_token"
return self.async_show_form(
step_id="auth",
data_schema=vol.Schema(
{
vol.Required(CONF_CREATE_TOKEN): bool,
vol.Optional(CONF_TOKEN): str,
}
),
errors=errors,
)
async def async_step_create_token(
self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]:
"""Send a request for a new token."""
if user_input is None:
self._auth_id = client.generate_random_auth_id()
return self.async_show_form(
step_id="create_token",
description_placeholders={
CONF_AUTH_ID: self._auth_id,
},
)
# Cancel the request token task if it's already running, then re-create it.
await self._cancel_request_token_task()
# Start a task in the background requesting a new token. The next step will
# wait on the response (which includes the user needing to visit the Hyperion
# UI to approve the request for a new token).
assert self.hass
assert self._auth_id is not None
self._request_token_task = self.hass.async_create_task(
self._request_token_task_func(self._auth_id)
)
return self.async_external_step(
step_id="create_token_external", url=self._get_hyperion_url()
)
async def async_step_create_token_external(
self, auth_resp: Optional[ConfigType] = None
) -> Dict[str, Any]:
"""Handle completion of the request for a new token."""
if auth_resp is not None and client.ResponseOK(auth_resp):
token = auth_resp.get(const.KEY_INFO, {}).get(const.KEY_TOKEN)
if token:
self._data[CONF_TOKEN] = token
return self.async_external_step_done(
next_step_id="create_token_success"
)
return self.async_external_step_done(next_step_id="create_token_fail")
async def async_step_create_token_success(
self, _: Optional[ConfigType] = None
) -> Dict[str, Any]:
"""Create an entry after successful token creation."""
# Clean-up the request task.
await self._cancel_request_token_task()
# Test the token.
login_ok = await self._can_login()
if login_ok is None:
return self.async_abort(reason="cannot_connect")
if not login_ok:
return self.async_abort(reason="auth_new_token_not_work_error")
return await self.async_step_confirm()
async def async_step_create_token_fail(
self, _: Optional[ConfigType] = None
) -> Dict[str, Any]:
"""Show an error on the auth form."""
# Clean-up the request task.
await self._cancel_request_token_task()
return self.async_abort(reason="auth_new_token_not_granted_error")
async def async_step_confirm(
self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]:
"""Get final confirmation before entry creation."""
if user_input is None and self._require_confirm:
return self.async_show_form(
step_id="confirm",
description_placeholders={
CONF_HOST: self._data[CONF_HOST],
CONF_PORT: self._data[CONF_PORT],
CONF_ID: self.unique_id,
},
)
async with self._create_client() as hyperion_client:
if not hyperion_client:
return self.async_abort(reason="cannot_connect")
hyperion_id = await hyperion_client.async_sysinfo_id()
if not hyperion_id:
return self.async_abort(reason="no_id")
await self.async_set_unique_id(hyperion_id, raise_on_progress=False)
self._abort_if_unique_id_configured()
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
return self.async_create_entry(
title=f"{self._data[CONF_HOST]}:{self._data[CONF_PORT]}", data=self._data
)
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> HyperionOptionsFlow:
"""Get the Hyperion Options flow."""
return HyperionOptionsFlow(config_entry)
class HyperionOptionsFlow(OptionsFlow):
"""Hyperion options flow."""
def __init__(self, config_entry: ConfigEntry):
"""Initialize a Hyperion options flow."""
self._config_entry = config_entry
async def async_step_init(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Optional(
CONF_PRIORITY,
default=self._config_entry.options.get(
CONF_PRIORITY, DEFAULT_PRIORITY
),
): vol.All(vol.Coerce(int), vol.Range(min=0, max=255)),
}
),
)

24
homeassistant/components/hyperion/const.py Normal file
View file

@ -0,0 +1,24 @@
"""Constants for Hyperion integration."""
DOMAIN = "hyperion"
DEFAULT_NAME = "Hyperion"
DEFAULT_ORIGIN = "Home Assistant"
DEFAULT_PRIORITY = 128
CONF_AUTH_ID = "auth_id"
CONF_CREATE_TOKEN = "create_token"
CONF_INSTANCE = "instance"
CONF_PRIORITY = "priority"
CONF_ROOT_CLIENT = "ROOT_CLIENT"
CONF_ON_UNLOAD = "ON_UNLOAD"
SIGNAL_INSTANCES_UPDATED = f"{DOMAIN}_instances_updated_signal." "{}"
SIGNAL_INSTANCE_REMOVED = f"{DOMAIN}_instance_removed_signal." "{}"
SOURCE_IMPORT = "import"
HYPERION_VERSION_WARN_CUTOFF = "2.0.0-alpha.9"
HYPERION_RELEASES_URL = "https://github.com/hyperion-project/hyperion.ng/releases"
TYPE_HYPERION_LIGHT = "hyperion_light"

339
homeassistant/components/hyperion/light.py
View file

@ -1,28 +1,59 @@
"""Support for Hyperion-NG remotes."""
from __future__ import annotations
import logging
import re
from types import MappingProxyType
from typing import Any, Callable, Dict, List, Optional, Sequence, Set, Tuple, cast
from hyperion import client, const
import voluptuous as vol
from homeassistant import data_entry_flow
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_EFFECT,
ATTR_HS_COLOR,
DOMAIN as LIGHT_DOMAIN,
PLATFORM_SCHEMA,
SUPPORT_BRIGHTNESS,
SUPPORT_COLOR,
SUPPORT_EFFECT,
LightEntity,
)
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, CONF_TOKEN
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
)
from homeassistant.helpers.entity_registry import async_get_registry
from homeassistant.helpers.typing import (
ConfigType,
DiscoveryInfoType,
HomeAssistantType,
)
import homeassistant.util.color as color_util
from . import async_create_connect_hyperion_client, get_hyperion_unique_id
from .const import (
CONF_ON_UNLOAD,
CONF_PRIORITY,
CONF_ROOT_CLIENT,
DEFAULT_ORIGIN,
DEFAULT_PRIORITY,
DOMAIN,
SIGNAL_INSTANCE_REMOVED,
SIGNAL_INSTANCES_UPDATED,
SOURCE_IMPORT,
TYPE_HYPERION_LIGHT,
)
_LOGGER = logging.getLogger(__name__)
CONF_DEFAULT_COLOR = "default_color"
CONF_PRIORITY = "priority"
CONF_HDMI_PRIORITY = "hdmi_priority"
CONF_EFFECT_LIST = "effect_list"
@ -35,21 +66,26 @@ CONF_EFFECT_LIST = "effect_list"
# showing a solid color. This is the same method used by WLED.
KEY_EFFECT_SOLID = "Solid"
KEY_ENTRY_ID_YAML = "YAML"
DEFAULT_COLOR = [255, 255, 255]
DEFAULT_BRIGHTNESS = 255
DEFAULT_EFFECT = KEY_EFFECT_SOLID
DEFAULT_NAME = "Hyperion"
DEFAULT_ORIGIN = "Home Assistant"
DEFAULT_PORT = 19444
DEFAULT_PRIORITY = 128
DEFAULT_PORT = const.DEFAULT_PORT_JSON
DEFAULT_HDMI_PRIORITY = 880
DEFAULT_EFFECT_LIST = []
DEFAULT_EFFECT_LIST: List[str] = []
SUPPORT_HYPERION = SUPPORT_COLOR | SUPPORT_BRIGHTNESS | SUPPORT_EFFECT
# Usage of YAML for configuration of the Hyperion component is deprecated.
PLATFORM_SCHEMA = vol.All(
cv.deprecated(CONF_HDMI_PRIORITY, invalidation_version="0.118"),
cv.deprecated(CONF_HOST),
cv.deprecated(CONF_PORT),
cv.deprecated(CONF_DEFAULT_COLOR, invalidation_version="0.118"),
cv.deprecated(CONF_NAME),
cv.deprecated(CONF_PRIORITY),
cv.deprecated(CONF_EFFECT_LIST, invalidation_version="0.118"),
PLATFORM_SCHEMA.extend(
{
@ -77,96 +113,277 @@ ICON_EFFECT = "mdi:lava-lamp"
ICON_EXTERNAL_SOURCE = "mdi:television-ambient-light"
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up a Hyperion server remote."""
name = config[CONF_NAME]
async def async_setup_platform(
hass: HomeAssistantType,
config: ConfigType,
async_add_entities: Callable,
discovery_info: Optional[DiscoveryInfoType] = None,
) -> None:
"""Set up Hyperion platform.."""
# This is the entrypoint for the old YAML-style Hyperion integration. The goal here
# is to auto-convert the YAML configuration into a config entry, with no human
# interaction, preserving the entity_id. This should be possible, as the YAML
# configuration did not support any of the things that should otherwise require
# human interaction in the config flow (e.g. it did not support auth).
host = config[CONF_HOST]
port = config[CONF_PORT]
priority = config[CONF_PRIORITY]
instance = 0 # YAML only supports a single instance.
hyperion_client = client.HyperionClient(host, port)
if not await hyperion_client.async_client_connect():
# First, connect to the server and get the server id (which will be unique_id on a config_entry
# if there is one).
hyperion_client = await async_create_connect_hyperion_client(host, port)
if not hyperion_client:
raise PlatformNotReady
hyperion_id = await hyperion_client.async_sysinfo_id()
if not hyperion_id:
raise PlatformNotReady
async_add_entities([Hyperion(name, priority, hyperion_client)])
future_unique_id = get_hyperion_unique_id(
hyperion_id, instance, TYPE_HYPERION_LIGHT
)
# Possibility 1: Already converted.
# There is already a config entry with the unique id reporting by the
# server. Nothing to do here.
for entry in hass.config_entries.async_entries(domain=DOMAIN):
if entry.unique_id == hyperion_id:
return
# Possibility 2: Upgraded to the new Hyperion component pre-config-flow.
# No config entry for this unique_id, but have an entity_registry entry
# with an old-style unique_id:
# <host>:<port>-<instance> (instance will always be 0, as YAML
# configuration does not support multiple
# instances)
# The unique_id needs to be updated, then the config_flow should do the rest.
registry = await async_get_registry(hass)
for entity_id, entity in registry.entities.items():
if entity.config_entry_id is not None or entity.platform != DOMAIN:
continue
result = re.search(rf"([^:]+):(\d+)-{instance}", entity.unique_id)
if result and result.group(1) == host and int(result.group(2)) == port:
registry.async_update_entity(entity_id, new_unique_id=future_unique_id)
break
else:
# Possibility 3: This is the first upgrade to the new Hyperion component.
# No config entry and no entity_registry entry, in which case the CONF_NAME
# variable will be used as the preferred name. Rather than pollute the config
# entry with a "suggested name" type variable, instead create an entry in the
# registry that will subsequently be used when the entity is created with this
# unique_id.
# This also covers the case that should not occur in the wild (no config entry,
# but new style unique_id).
registry.async_get_or_create(
domain=LIGHT_DOMAIN,
platform=DOMAIN,
unique_id=future_unique_id,
suggested_object_id=config[CONF_NAME],
)
async def migrate_yaml_to_config_entry_and_options(
host: str, port: int, priority: int
) -> None:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_HOST: host,
CONF_PORT: port,
},
)
if (
result["type"] != data_entry_flow.RESULT_TYPE_CREATE_ENTRY
or result.get("result") is None
):
_LOGGER.warning(
"Could not automatically migrate Hyperion YAML to a config entry."
)
return
config_entry = result.get("result")
options = {**config_entry.options, CONF_PRIORITY: config[CONF_PRIORITY]}
hass.config_entries.async_update_entry(config_entry, options=options)
_LOGGER.info(
"Successfully migrated Hyperion YAML configuration to a config entry."
)
# Kick off a config flow to create the config entry.
hass.async_create_task(
migrate_yaml_to_config_entry_and_options(host, port, config[CONF_PRIORITY])
)
class Hyperion(LightEntity):
async def async_setup_entry(
hass: HomeAssistantType, config_entry: ConfigEntry, async_add_entities: Callable
) -> bool:
"""Set up a Hyperion platform from config entry."""
host = config_entry.data[CONF_HOST]
port = config_entry.data[CONF_PORT]
token = config_entry.data.get(CONF_TOKEN)
async def async_instances_to_entities(response: Dict[str, Any]) -> None:
if not response or const.KEY_DATA not in response:
return
await async_instances_to_entities_raw(response[const.KEY_DATA])
async def async_instances_to_entities_raw(instances: List[Dict[str, Any]]) -> None:
registry = await async_get_registry(hass)
entities_to_add: List[HyperionLight] = []
desired_unique_ids: Set[str] = set()
server_id = cast(str, config_entry.unique_id)
# In practice, an instance can be in 3 states as seen by this function:
#
# * Exists, and is running: Add it to hass.
# * Exists, but is not running: Cannot add yet, but should not delete it either.
# It will show up as "unavailable".
# * No longer exists: Delete it from hass.
# Add instances that are missing.
for instance in instances:
instance_id = instance.get(const.KEY_INSTANCE)
if instance_id is None or not instance.get(const.KEY_RUNNING, False):
continue
unique_id = get_hyperion_unique_id(
server_id, instance_id, TYPE_HYPERION_LIGHT
)
desired_unique_ids.add(unique_id)
if unique_id in current_entities:
continue
hyperion_client = await async_create_connect_hyperion_client(
host, port, instance=instance_id, token=token
)
if not hyperion_client:
continue
current_entities.add(unique_id)
entities_to_add.append(
HyperionLight(
unique_id,
instance.get(const.KEY_FRIENDLY_NAME, DEFAULT_NAME),
config_entry.options,
hyperion_client,
)
)
# Delete instances that are no longer present on this server.
for unique_id in current_entities - desired_unique_ids:
current_entities.remove(unique_id)
async_dispatcher_send(hass, SIGNAL_INSTANCE_REMOVED.format(unique_id))
entity_id = registry.async_get_entity_id(LIGHT_DOMAIN, DOMAIN, unique_id)
if entity_id:
registry.async_remove(entity_id)
async_add_entities(entities_to_add)
# Readability note: This variable is kept alive in the context of the callback to
# async_instances_to_entities below.
current_entities: Set[str] = set()
await async_instances_to_entities_raw(
hass.data[DOMAIN][config_entry.entry_id][CONF_ROOT_CLIENT].instances,
)
hass.data[DOMAIN][config_entry.entry_id][CONF_ON_UNLOAD].append(
async_dispatcher_connect(
hass,
SIGNAL_INSTANCES_UPDATED.format(config_entry.entry_id),
async_instances_to_entities,
)
)
return True
class HyperionLight(LightEntity):
"""Representation of a Hyperion remote."""
def __init__(self, name, priority, hyperion_client):
def __init__(
self,
unique_id: str,
name: str,
options: MappingProxyType[str, Any],
hyperion_client: client.HyperionClient,
) -> None:
"""Initialize the light."""
self._unique_id = unique_id
self._name = name
self._priority = priority
self._options = options
self._client = hyperion_client
# Active state representing the Hyperion instance.
self._set_internal_state(
brightness=255, rgb_color=DEFAULT_COLOR, effect=KEY_EFFECT_SOLID
)
self._effect_list = []
self._brightness: int = 255
self._rgb_color: Sequence[int] = DEFAULT_COLOR
self._effect: str = KEY_EFFECT_SOLID
self._icon: str = ICON_LIGHTBULB
self._effect_list: List[str] = []
@property
def should_poll(self):
def should_poll(self) -> bool:
"""Return whether or not this entity should be polled."""
return False
@property
def name(self):
def name(self) -> str:
"""Return the name of the light."""
return self._name
@property
def brightness(self):
def brightness(self) -> int:
"""Return the brightness of this light between 0..255."""
return self._brightness
@property
def hs_color(self):
def hs_color(self) -> Tuple[float, float]:
"""Return last color value set."""
return color_util.color_RGB_to_hs(*self._rgb_color)
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if not black."""
return self._client.is_on()
return bool(self._client.is_on())
@property
def icon(self):
def icon(self) -> str:
"""Return state specific icon."""
return self._icon
@property
def effect(self):
def effect(self) -> str:
"""Return the current effect."""
return self._effect
@property
def effect_list(self):
def effect_list(self) -> List[str]:
"""Return the list of supported effects."""
return (
self._effect_list
+ const.KEY_COMPONENTID_EXTERNAL_SOURCES
+ list(const.KEY_COMPONENTID_EXTERNAL_SOURCES)
+ [KEY_EFFECT_SOLID]
)
@property
def supported_features(self):
def supported_features(self) -> int:
"""Flag supported features."""
return SUPPORT_HYPERION
@property
def available(self):
def available(self) -> bool:
"""Return server availability."""
return self._client.has_loaded_state
return bool(self._client.has_loaded_state)
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return a unique id for this instance."""
return self._client.id
return self._unique_id
async def async_turn_on(self, **kwargs):
def _get_option(self, key: str) -> Any:
"""Get a value from the provided options."""
defaults = {CONF_PRIORITY: DEFAULT_PRIORITY}
return self._options.get(key, defaults[key])
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the lights on."""
# == Turn device on ==
# Turn on both ALL (Hyperion itself) and LEDDEVICE. It would be
@ -197,6 +414,7 @@ class Hyperion(LightEntity):
# == Get key parameters ==
brightness = kwargs.get(ATTR_BRIGHTNESS, self._brightness)
effect = kwargs.get(ATTR_EFFECT, self._effect)
rgb_color: Sequence[int]
if ATTR_HS_COLOR in kwargs:
rgb_color = color_util.color_hs_to_RGB(*kwargs[ATTR_HS_COLOR])
else:
@ -220,7 +438,7 @@ class Hyperion(LightEntity):
# Clear any color/effect.
if not await self._client.async_send_clear(
**{const.KEY_PRIORITY: self._priority}
**{const.KEY_PRIORITY: self._get_option(CONF_PRIORITY)}
):
return
@ -241,13 +459,13 @@ class Hyperion(LightEntity):
# This call should not be necessary, but without it there is no priorities-update issued:
# https://github.com/hyperion-project/hyperion.ng/issues/992
if not await self._client.async_send_clear(
**{const.KEY_PRIORITY: self._priority}
**{const.KEY_PRIORITY: self._get_option(CONF_PRIORITY)}
):
return
if not await self._client.async_send_set_effect(
**{
const.KEY_PRIORITY: self._priority,
const.KEY_PRIORITY: self._get_option(CONF_PRIORITY),
const.KEY_EFFECT: {const.KEY_NAME: effect},
const.KEY_ORIGIN: DEFAULT_ORIGIN,
}
@ -257,14 +475,14 @@ class Hyperion(LightEntity):
else:
if not await self._client.async_send_set_color(
**{
const.KEY_PRIORITY: self._priority,
const.KEY_PRIORITY: self._get_option(CONF_PRIORITY),
const.KEY_COLOR: rgb_color,
const.KEY_ORIGIN: DEFAULT_ORIGIN,
}
):
return
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Disable the LED output component."""
if not await self._client.async_send_set_component(
**{
@ -276,7 +494,12 @@ class Hyperion(LightEntity):
):
return
def _set_internal_state(self, brightness=None, rgb_color=None, effect=None):
def _set_internal_state(
self,
brightness: Optional[int] = None,
rgb_color: Optional[Sequence[int]] = None,
effect: Optional[str] = None,
) -> None:
"""Set the internal state."""
if brightness is not None:
self._brightness = brightness
@ -291,11 +514,11 @@ class Hyperion(LightEntity):
else:
self._icon = ICON_EFFECT
def _update_components(self, _=None):
def _update_components(self, _: Optional[Dict[str, Any]] = None) -> None:
"""Update Hyperion components."""
self.async_write_ha_state()
def _update_adjustment(self, _=None):
def _update_adjustment(self, _: Optional[Dict[str, Any]] = None) -> None:
"""Update Hyperion adjustments."""
if self._client.adjustment:
brightness_pct = self._client.adjustment[0].get(
@ -308,7 +531,7 @@ class Hyperion(LightEntity):
)
self.async_write_ha_state()
def _update_priorities(self, _=None):
def _update_priorities(self, _: Optional[Dict[str, Any]] = None) -> None:
"""Update Hyperion priorities."""
visible_priority = self._client.visible_priority
if visible_priority:
@ -328,11 +551,11 @@ class Hyperion(LightEntity):
)
self.async_write_ha_state()
def _update_effect_list(self, _=None):
def _update_effect_list(self, _: Optional[Dict[str, Any]] = None) -> None:
"""Update Hyperion effects."""
if not self._client.effects:
return
effect_list = []
effect_list: List[str] = []
for effect in self._client.effects or []:
if const.KEY_NAME in effect:
effect_list.append(effect[const.KEY_NAME])
@ -340,7 +563,7 @@ class Hyperion(LightEntity):
self._effect_list = effect_list
self.async_write_ha_state()
def _update_full_state(self):
def _update_full_state(self) -> None:
"""Update full Hyperion state."""
self._update_adjustment()
self._update_priorities()
@ -356,12 +579,21 @@ class Hyperion(LightEntity):
self._rgb_color,
)
def _update_client(self, json):
def _update_client(self, _: Optional[Dict[str, Any]] = None) -> None:
"""Update client connection state."""
self.async_write_ha_state()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Register callbacks when entity added to hass."""
assert self.hass
self.async_on_remove(
async_dispatcher_connect(
self.hass,
SIGNAL_INSTANCE_REMOVED.format(self._unique_id),
self.async_remove,
)
)
self._client.set_callbacks(
{
f"{const.KEY_ADJUSTMENT}-{const.KEY_UPDATE}": self._update_adjustment,
@ -374,4 +606,7 @@ class Hyperion(LightEntity):
# Load initial state.
self._update_full_state()
return True
async def async_will_remove_from_hass(self) -> None:
"""Disconnect from server."""
await self._client.async_client_disconnect()

19
homeassistant/components/hyperion/manifest.json
View file

@ -1,7 +1,18 @@
{
"codeowners": [
"@dermotduffy"
],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/hyperion",
"domain": "hyperion",
"name": "Hyperion",
"documentation": "https://www.home-assistant.io/integrations/hyperion",
"requirements": ["hyperion-py==0.3.0"],
"codeowners": ["@dermotduffy"]
}
"requirements": [
"hyperion-py==0.6.0"
],
"ssdp": [
{
"manufacturer": "Hyperion Open Source Ambient Lighting",
"st": "urn:hyperion-project.org:device:basic:1"
}
]
}

52
homeassistant/components/hyperion/strings.json Normal file
View file

@ -0,0 +1,52 @@
{
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]"
}
},
"auth": {
"description": "Configure authorization to your Hyperion Ambilight server",
"data": {
"create_token": "Automatically create new token",
"token": "Or provide pre-existing token"
}
},
"create_token": {
"description": "Choose **Submit** below to request a new authentication token. You will be redirected to the Hyperion UI to approve the request. Please verify the shown id is \"{auth_id}\"",
"title": "Automatically create new authentication token"
},
"create_token_external": {
"title": "Accept new token in Hyperion UI"
},
"confirm": {
"description": "Do you want to add this Hyperion Ambilight to Home Assistant?\n\n**Host:** {host}\n**Port:** {port}\n**ID**: {id}",
"title": "Confirm addition of Hyperion Ambilight service"
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_access_token": "[%key:common::config_flow::error::invalid_access_token%]"
},
"abort": {
"auth_required_error": "Failed to determine if authorization is required",
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"auth_new_token_not_granted_error": "Newly created token was not approved on Hyperion UI",
"auth_new_token_not_work_error": "Failed to authenticate using newly created token",
"no_id": "The Hyperion Ambilight instance did not report its id"
}
},
"options": {
"step": {
"init": {
"data": {
"priority": "Hyperion priority to use for colors and effects"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View file

@ -92,6 +92,7 @@ FLOWS = [
"hue",
"hunterdouglas_powerview",
"hvv_departures",
"hyperion",
"iaqualink",
"icloud",
"ifttt",

6
homeassistant/generated/ssdp.py
View file

@ -114,6 +114,12 @@ SSDP = {
"modelName": "Philips hue bridge 2015"
}
],
"hyperion": [
{
"manufacturer": "Hyperion Open Source Ambient Lighting",
"st": "urn:hyperion-project.org:device:basic:1"
}
],
"isy994": [
{
"deviceType": "urn:udi-com:device:X_Insteon_Lighting_Device:1",

2
requirements_all.txt
View file

@ -790,7 +790,7 @@ huawei-lte-api==1.4.12
hydrawiser==0.2
# homeassistant.components.hyperion
hyperion-py==0.3.0
hyperion-py==0.6.0
# homeassistant.components.bh1750
# homeassistant.components.bme280

2
requirements_test_all.txt
View file

@ -413,7 +413,7 @@ httplib2==0.10.3
huawei-lte-api==1.4.12
# homeassistant.components.hyperion
hyperion-py==0.3.0
hyperion-py==0.6.0
# homeassistant.components.iaqualink
iaqualink==0.3.4

2
setup.cfg
View file

@ -40,7 +40,7 @@ warn_incomplete_stub = true
warn_redundant_casts = true
warn_unused_configs = true
[mypy-homeassistant.block_async_io,homeassistant.bootstrap,homeassistant.components,homeassistant.config_entries,homeassistant.config,homeassistant.const,homeassistant.core,homeassistant.data_entry_flow,homeassistant.exceptions,homeassistant.__init__,homeassistant.loader,homeassistant.__main__,homeassistant.requirements,homeassistant.runner,homeassistant.setup,homeassistant.util,homeassistant.auth.*,homeassistant.components.automation.*,homeassistant.components.binary_sensor.*,homeassistant.components.calendar.*,homeassistant.components.cover.*,homeassistant.components.device_automation.*,homeassistant.components.frontend.*,homeassistant.components.geo_location.*,homeassistant.components.group.*,homeassistant.components.history.*,homeassistant.components.http.*,homeassistant.components.huawei_lte.*,homeassistant.components.image_processing.*,homeassistant.components.integration.*,homeassistant.components.light.*,homeassistant.components.lock.*,homeassistant.components.mailbox.*,homeassistant.components.media_player.*,homeassistant.components.notify.*,homeassistant.components.persistent_notification.*,homeassistant.components.proximity.*,homeassistant.components.remote.*,homeassistant.components.scene.*,homeassistant.components.sensor.*,homeassistant.components.sun.*,homeassistant.components.switch.*,homeassistant.components.systemmonitor.*,homeassistant.components.tts.*,homeassistant.components.vacuum.*,homeassistant.components.water_heater.*,homeassistant.components.weather.*,homeassistant.components.websocket_api.*,homeassistant.components.zone.*,homeassistant.helpers.*,homeassistant.scripts.*,homeassistant.util.*]
[mypy-homeassistant.block_async_io,homeassistant.bootstrap,homeassistant.components,homeassistant.config_entries,homeassistant.config,homeassistant.const,homeassistant.core,homeassistant.data_entry_flow,homeassistant.exceptions,homeassistant.__init__,homeassistant.loader,homeassistant.__main__,homeassistant.requirements,homeassistant.runner,homeassistant.setup,homeassistant.util,homeassistant.auth.*,homeassistant.components.automation.*,homeassistant.components.binary_sensor.*,homeassistant.components.calendar.*,homeassistant.components.cover.*,homeassistant.components.device_automation.*,homeassistant.components.frontend.*,homeassistant.components.geo_location.*,homeassistant.components.group.*,homeassistant.components.history.*,homeassistant.components.http.*,homeassistant.components.huawei_lte.*,homeassistant.components.hyperion.*,homeassistant.components.image_processing.*,homeassistant.components.integration.*,homeassistant.components.light.*,homeassistant.components.lock.*,homeassistant.components.mailbox.*,homeassistant.components.media_player.*,homeassistant.components.notify.*,homeassistant.components.persistent_notification.*,homeassistant.components.proximity.*,homeassistant.components.remote.*,homeassistant.components.scene.*,homeassistant.components.sensor.*,homeassistant.components.sun.*,homeassistant.components.switch.*,homeassistant.components.systemmonitor.*,homeassistant.components.tts.*,homeassistant.components.vacuum.*,homeassistant.components.water_heater.*,homeassistant.components.weather.*,homeassistant.components.websocket_api.*,homeassistant.components.zone.*,homeassistant.helpers.*,homeassistant.scripts.*,homeassistant.util.*,tests.components.hyperion.*]
strict = true
ignore_errors = false
warn_unreachable = true

135
tests/components/hyperion/__init__.py
View file

@ -1 +1,136 @@
"""Tests for the Hyperion component."""
from __future__ import annotations
import logging
from types import TracebackType
from typing import Any, Dict, Optional, Type
from hyperion import const
from homeassistant.components.hyperion.const import CONF_PRIORITY, DOMAIN
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.helpers.typing import HomeAssistantType
from tests.async_mock import AsyncMock, Mock, patch # type: ignore[attr-defined]
from tests.common import MockConfigEntry
TEST_HOST = "test"
TEST_PORT = const.DEFAULT_PORT_JSON + 1
TEST_PORT_UI = const.DEFAULT_PORT_UI + 1
TEST_INSTANCE = 1
TEST_SYSINFO_ID = "f9aab089-f85a-55cf-b7c1-222a72faebe9"
TEST_SYSINFO_VERSION = "2.0.0-alpha.8"
TEST_PRIORITY = 180
TEST_YAML_NAME = f"{TEST_HOST}_{TEST_PORT}_{TEST_INSTANCE}"
TEST_YAML_ENTITY_ID = f"{LIGHT_DOMAIN}.{TEST_YAML_NAME}"
TEST_ENTITY_ID_1 = "light.test_instance_1"
TEST_ENTITY_ID_2 = "light.test_instance_2"
TEST_ENTITY_ID_3 = "light.test_instance_3"
TEST_TITLE = f"{TEST_HOST}:{TEST_PORT}"
TEST_TOKEN = "sekr1t"
TEST_CONFIG_ENTRY_ID = "74565ad414754616000674c87bdc876c"
TEST_CONFIG_ENTRY_OPTIONS: Dict[str, Any] = {CONF_PRIORITY: TEST_PRIORITY}
TEST_INSTANCE_1: Dict[str, Any] = {
"friendly_name": "Test instance 1",
"instance": 1,
"running": True,
}
TEST_INSTANCE_2: Dict[str, Any] = {
"friendly_name": "Test instance 2",
"instance": 2,
"running": True,
}
TEST_INSTANCE_3: Dict[str, Any] = {
"friendly_name": "Test instance 3",
"instance": 3,
"running": True,
}
_LOGGER = logging.getLogger(__name__)
class AsyncContextManagerMock(Mock): # type: ignore[misc]
"""An async context manager mock for Hyperion."""
async def __aenter__(self) -> Optional[AsyncContextManagerMock]:
"""Enter context manager and connect the client."""
result = await self.async_client_connect()
return self if result else None
async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
"""Leave context manager and disconnect the client."""
await self.async_client_disconnect()
def create_mock_client() -> Mock:
"""Create a mock Hyperion client."""
mock_client = AsyncContextManagerMock()
# pylint: disable=attribute-defined-outside-init
mock_client.async_client_connect = AsyncMock(return_value=True)
mock_client.async_client_disconnect = AsyncMock(return_value=True)
mock_client.async_is_auth_required = AsyncMock(
return_value={
"command": "authorize-tokenRequired",
"info": {"required": False},
"success": True,
"tan": 1,
}
)
mock_client.async_login = AsyncMock(
return_value={"command": "authorize-login", "success": True, "tan": 0}
)
mock_client.async_sysinfo_id = AsyncMock(return_value=TEST_SYSINFO_ID)
mock_client.async_sysinfo_version = AsyncMock(return_value=TEST_SYSINFO_ID)
mock_client.adjustment = None
mock_client.effects = None
mock_client.instances = [
{"friendly_name": "Test instance 1", "instance": 0, "running": True}
]
return mock_client
def add_test_config_entry(hass: HomeAssistantType) -> ConfigEntry:
"""Add a test config entry."""
config_entry: MockConfigEntry = MockConfigEntry( # type: ignore[no-untyped-call]
entry_id=TEST_CONFIG_ENTRY_ID,
domain=DOMAIN,
data={
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
},
title=f"Hyperion {TEST_SYSINFO_ID}",
unique_id=TEST_SYSINFO_ID,
options=TEST_CONFIG_ENTRY_OPTIONS,
)
config_entry.add_to_hass(hass) # type: ignore[no-untyped-call]
return config_entry
async def setup_test_config_entry(
hass: HomeAssistantType, hyperion_client: Optional[Mock] = None
) -> ConfigEntry:
"""Add a test Hyperion entity to hass."""
config_entry = add_test_config_entry(hass)
hyperion_client = hyperion_client or create_mock_client()
# pylint: disable=attribute-defined-outside-init
hyperion_client.instances = [TEST_INSTANCE_1]
with patch(
"homeassistant.components.hyperion.client.HyperionClient",
return_value=hyperion_client,
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry

696
tests/components/hyperion/test_config_flow.py Normal file
View file

@ -0,0 +1,696 @@
"""Tests for the Hyperion config flow."""
import logging
from typing import Any, Dict, Optional
from hyperion import const
from homeassistant import data_entry_flow
from homeassistant.components.hyperion.const import (
CONF_AUTH_ID,
CONF_CREATE_TOKEN,
CONF_PRIORITY,
DOMAIN,
SOURCE_IMPORT,
)
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_HOST,
CONF_PORT,
CONF_TOKEN,
SERVICE_TURN_ON,
)
from homeassistant.helpers.typing import HomeAssistantType
from . import (
TEST_CONFIG_ENTRY_ID,
TEST_ENTITY_ID_1,
TEST_HOST,
TEST_INSTANCE,
TEST_PORT,
TEST_PORT_UI,
TEST_SYSINFO_ID,
TEST_TITLE,
TEST_TOKEN,
add_test_config_entry,
create_mock_client,
)
from tests.async_mock import AsyncMock, patch # type: ignore[attr-defined]
from tests.common import MockConfigEntry
_LOGGER = logging.getLogger(__name__)
TEST_IP_ADDRESS = "192.168.0.1"
TEST_HOST_PORT: Dict[str, Any] = {
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
}
TEST_AUTH_REQUIRED_RESP = {
"command": "authorize-tokenRequired",
"info": {
"required": True,
},
"success": True,
"tan": 1,
}
TEST_AUTH_ID = "ABCDE"
TEST_REQUEST_TOKEN_SUCCESS = {
"command": "authorize-requestToken",
"success": True,
"info": {"comment": const.DEFAULT_ORIGIN, "id": TEST_AUTH_ID, "token": TEST_TOKEN},
}
TEST_REQUEST_TOKEN_FAIL = {
"command": "authorize-requestToken",
"success": False,
"error": "Token request timeout or denied",
}
TEST_SSDP_SERVICE_INFO = {
"ssdp_location": f"http://{TEST_HOST}:{TEST_PORT_UI}/description.xml",
"ssdp_st": "upnp:rootdevice",
"deviceType": "urn:schemas-upnp-org:device:Basic:1",
"friendlyName": f"Hyperion ({TEST_HOST})",
"manufacturer": "Hyperion Open Source Ambient Lighting",
"manufacturerURL": "https://www.hyperion-project.org",
"modelDescription": "Hyperion Open Source Ambient Light",
"modelName": "Hyperion",
"modelNumber": "2.0.0-alpha.8",
"modelURL": "https://www.hyperion-project.org",
"serialNumber": f"{TEST_SYSINFO_ID}",
"UDN": f"uuid:{TEST_SYSINFO_ID}",
"ports": {
"jsonServer": f"{TEST_PORT}",
"sslServer": "8092",
"protoBuffer": "19445",
"flatBuffer": "19400",
},
"presentationURL": "index.html",
"iconList": {
"icon": {
"mimetype": "image/png",
"height": "100",
"width": "100",
"depth": "32",
"url": "img/hyperion/ssdp_icon.png",
}
},
"ssdp_usn": f"uuid:{TEST_SYSINFO_ID}",
"ssdp_ext": "",
"ssdp_server": "Raspbian GNU/Linux 10 (buster)/10 UPnP/1.0 Hyperion/2.0.0-alpha.8",
}
async def _create_mock_entry(hass: HomeAssistantType) -> MockConfigEntry:
"""Add a test Hyperion entity to hass."""
entry: MockConfigEntry = MockConfigEntry( # type: ignore[no-untyped-call]
entry_id=TEST_CONFIG_ENTRY_ID,
domain=DOMAIN,
unique_id=TEST_SYSINFO_ID,
title=TEST_TITLE,
data={
"host": TEST_HOST,
"port": TEST_PORT,
"instance": TEST_INSTANCE,
},
)
entry.add_to_hass(hass) # type: ignore[no-untyped-call]
# Setup
client = create_mock_client()
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
async def _init_flow(
hass: HomeAssistantType,
source: str = SOURCE_USER,
data: Optional[Dict[str, Any]] = None,
) -> Any:
"""Initialize a flow."""
data = data or {}
return await hass.config_entries.flow.async_init(
DOMAIN, context={"source": source}, data=data
)
async def _configure_flow(
hass: HomeAssistantType, result: Dict, user_input: Optional[Dict[str, Any]] = None
) -> Any:
"""Provide input to a flow."""
user_input = user_input or {}
with patch(
"homeassistant.components.hyperion.async_setup", return_value=True
), patch(
"homeassistant.components.hyperion.async_setup_entry",
return_value=True,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=user_input
)
await hass.async_block_till_done()
return result
async def test_user_if_no_configuration(hass: HomeAssistantType) -> None:
"""Check flow behavior when no configuration is present."""
result = await _init_flow(hass)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["handler"] == DOMAIN
async def test_user_existing_id_abort(hass: HomeAssistantType) -> None:
"""Verify a duplicate ID results in an abort."""
result = await _init_flow(hass)
await _create_mock_entry(hass)
client = create_mock_client()
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_user_client_errors(hass: HomeAssistantType) -> None:
"""Verify correct behaviour with client errors."""
result = await _init_flow(hass)
client = create_mock_client()
# Fail the connection.
client.async_client_connect = AsyncMock(return_value=False)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"]["base"] == "cannot_connect"
# Fail the auth check call.
client.async_client_connect = AsyncMock(return_value=True)
client.async_is_auth_required = AsyncMock(return_value={"success": False})
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "auth_required_error"
async def test_user_confirm_cannot_connect(hass: HomeAssistantType) -> None:
"""Test a failure to connect during confirmation."""
result = await _init_flow(hass)
good_client = create_mock_client()
bad_client = create_mock_client()
bad_client.async_client_connect = AsyncMock(return_value=False)
# Confirmation sync_client_connect fails.
with patch(
"homeassistant.components.hyperion.client.HyperionClient",
side_effect=[good_client, bad_client],
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def test_user_confirm_id_error(hass: HomeAssistantType) -> None:
"""Test a failure fetching the server id during confirmation."""
result = await _init_flow(hass)
client = create_mock_client()
client.async_sysinfo_id = AsyncMock(return_value=None)
# Confirmation sync_client_connect fails.
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "no_id"
async def test_user_noauth_flow_success(hass: HomeAssistantType) -> None:
"""Check a full flow without auth."""
result = await _init_flow(hass)
client = create_mock_client()
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["handler"] == DOMAIN
assert result["title"] == TEST_TITLE
assert result["data"] == {
**TEST_HOST_PORT,
}
async def test_user_auth_required(hass: HomeAssistantType) -> None:
"""Verify correct behaviour when auth is required."""
result = await _init_flow(hass)
client = create_mock_client()
client.async_is_auth_required = AsyncMock(return_value=TEST_AUTH_REQUIRED_RESP)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
async def test_auth_static_token_auth_required_fail(hass: HomeAssistantType) -> None:
"""Verify correct behaviour with a failed auth required call."""
result = await _init_flow(hass)
client = create_mock_client()
client.async_is_auth_required = AsyncMock(return_value=None)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "auth_required_error"
async def test_auth_static_token_success(hass: HomeAssistantType) -> None:
"""Test a successful flow with a static token."""
result = await _init_flow(hass)
assert result["step_id"] == "user"
client = create_mock_client()
client.async_is_auth_required = AsyncMock(return_value=TEST_AUTH_REQUIRED_RESP)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
result = await _configure_flow(
hass, result, user_input={CONF_CREATE_TOKEN: False, CONF_TOKEN: TEST_TOKEN}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["handler"] == DOMAIN
assert result["title"] == TEST_TITLE
assert result["data"] == {
**TEST_HOST_PORT,
CONF_TOKEN: TEST_TOKEN,
}
async def test_auth_static_token_login_fail(hass: HomeAssistantType) -> None:
"""Test correct behavior with a bad static token."""
result = await _init_flow(hass)
assert result["step_id"] == "user"
client = create_mock_client()
client.async_is_auth_required = AsyncMock(return_value=TEST_AUTH_REQUIRED_RESP)
# Fail the login call.
client.async_login = AsyncMock(
return_value={"command": "authorize-login", "success": False, "tan": 0}
)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
result = await _configure_flow(
hass, result, user_input={CONF_CREATE_TOKEN: False, CONF_TOKEN: TEST_TOKEN}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"]["base"] == "invalid_access_token"
async def test_auth_create_token_approval_declined(hass: HomeAssistantType) -> None:
"""Verify correct behaviour when a token request is declined."""
result = await _init_flow(hass)
client = create_mock_client()
client.async_is_auth_required = AsyncMock(return_value=TEST_AUTH_REQUIRED_RESP)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
client.async_request_token = AsyncMock(return_value=TEST_REQUEST_TOKEN_FAIL)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
), patch(
"homeassistant.components.hyperion.config_flow.client.generate_random_auth_id",
return_value=TEST_AUTH_ID,
):
result = await _configure_flow(
hass, result, user_input={CONF_CREATE_TOKEN: True}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "create_token"
assert result["description_placeholders"] == {
CONF_AUTH_ID: TEST_AUTH_ID,
}
result = await _configure_flow(hass, result)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
assert result["step_id"] == "create_token_external"
# The flow will be automatically advanced by the auth token response.
result = await _configure_flow(hass, result)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "auth_new_token_not_granted_error"
async def test_auth_create_token_when_issued_token_fails(
hass: HomeAssistantType,
) -> None:
"""Verify correct behaviour when a token is granted by fails to authenticate."""
result = await _init_flow(hass)
client = create_mock_client()
client.async_is_auth_required = AsyncMock(return_value=TEST_AUTH_REQUIRED_RESP)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
client.async_request_token = AsyncMock(return_value=TEST_REQUEST_TOKEN_SUCCESS)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
), patch(
"homeassistant.components.hyperion.config_flow.client.generate_random_auth_id",
return_value=TEST_AUTH_ID,
):
result = await _configure_flow(
hass, result, user_input={CONF_CREATE_TOKEN: True}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "create_token"
assert result["description_placeholders"] == {
CONF_AUTH_ID: TEST_AUTH_ID,
}
result = await _configure_flow(hass, result)
assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
assert result["step_id"] == "create_token_external"
# The flow will be automatically advanced by the auth token response.
# Make the last verification fail.
client.async_client_connect = AsyncMock(return_value=False)
result = await _configure_flow(hass, result)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def test_auth_create_token_success(hass: HomeAssistantType) -> None:
"""Verify correct behaviour when a token is successfully created."""
result = await _init_flow(hass)
client = create_mock_client()
client.async_is_auth_required = AsyncMock(return_value=TEST_AUTH_REQUIRED_RESP)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result, user_input=TEST_HOST_PORT)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
client.async_request_token = AsyncMock(return_value=TEST_REQUEST_TOKEN_SUCCESS)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
), patch(
"homeassistant.components.hyperion.config_flow.client.generate_random_auth_id",
return_value=TEST_AUTH_ID,
):
result = await _configure_flow(
hass, result, user_input={CONF_CREATE_TOKEN: True}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "create_token"
assert result["description_placeholders"] == {
CONF_AUTH_ID: TEST_AUTH_ID,
}
result = await _configure_flow(hass, result)
assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
assert result["step_id"] == "create_token_external"
# The flow will be automatically advanced by the auth token response.
result = await _configure_flow(hass, result)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["handler"] == DOMAIN
assert result["title"] == TEST_TITLE
assert result["data"] == {
**TEST_HOST_PORT,
CONF_TOKEN: TEST_TOKEN,
}
async def test_ssdp_success(hass: HomeAssistantType) -> None:
"""Check an SSDP flow."""
client = create_mock_client()
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _init_flow(hass, source=SOURCE_SSDP, data=TEST_SSDP_SERVICE_INFO)
await hass.async_block_till_done()
# Accept the confirmation.
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _configure_flow(hass, result)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["handler"] == DOMAIN
assert result["title"] == TEST_TITLE
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
}
async def test_ssdp_cannot_connect(hass: HomeAssistantType) -> None:
"""Check an SSDP flow that cannot connect."""
client = create_mock_client()
client.async_client_connect = AsyncMock(return_value=False)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _init_flow(hass, source=SOURCE_SSDP, data=TEST_SSDP_SERVICE_INFO)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def test_ssdp_missing_serial(hass: HomeAssistantType) -> None:
"""Check an SSDP flow where no id is provided."""
client = create_mock_client()
bad_data = {**TEST_SSDP_SERVICE_INFO}
del bad_data["serialNumber"]
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _init_flow(hass, source=SOURCE_SSDP, data=bad_data)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "no_id"
async def test_ssdp_failure_bad_port_json(hass: HomeAssistantType) -> None:
"""Check an SSDP flow with bad json port."""
client = create_mock_client()
bad_data: Dict[str, Any] = {**TEST_SSDP_SERVICE_INFO}
bad_data["ports"]["jsonServer"] = "not_a_port"
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _init_flow(hass, source=SOURCE_SSDP, data=bad_data)
result = await _configure_flow(hass, result)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"][CONF_PORT] == const.DEFAULT_PORT_JSON
async def test_ssdp_failure_bad_port_ui(hass: HomeAssistantType) -> None:
"""Check an SSDP flow with bad ui port."""
client = create_mock_client()
client.async_is_auth_required = AsyncMock(return_value=TEST_AUTH_REQUIRED_RESP)
bad_data = {**TEST_SSDP_SERVICE_INFO}
bad_data["ssdp_location"] = f"http://{TEST_HOST}:not_a_port/description.xml"
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
), patch(
"homeassistant.components.hyperion.config_flow.client.generate_random_auth_id",
return_value=TEST_AUTH_ID,
):
result = await _init_flow(hass, source=SOURCE_SSDP, data=bad_data)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
client.async_request_token = AsyncMock(return_value=TEST_REQUEST_TOKEN_FAIL)
result = await _configure_flow(
hass, result, user_input={CONF_CREATE_TOKEN: True}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "create_token"
# Verify a working URL is used despite the bad port number
assert result["description_placeholders"] == {
CONF_AUTH_ID: TEST_AUTH_ID,
}
async def test_ssdp_abort_duplicates(hass: HomeAssistantType) -> None:
"""Check an SSDP flow where no id is provided."""
client = create_mock_client()
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result_1 = await _init_flow(
hass, source=SOURCE_SSDP, data=TEST_SSDP_SERVICE_INFO
)
result_2 = await _init_flow(
hass, source=SOURCE_SSDP, data=TEST_SSDP_SERVICE_INFO
)
await hass.async_block_till_done()
assert result_1["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result_2["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result_2["reason"] == "already_in_progress"
async def test_import_success(hass: HomeAssistantType) -> None:
"""Check an import flow from the old-style YAML."""
client = create_mock_client()
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _init_flow(
hass,
source=SOURCE_IMPORT,
data={
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
},
)
await hass.async_block_till_done()
# No human interaction should be required.
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["handler"] == DOMAIN
assert result["title"] == TEST_TITLE
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
}
async def test_import_cannot_connect(hass: HomeAssistantType) -> None:
"""Check an import flow that cannot connect."""
client = create_mock_client()
client.async_client_connect = AsyncMock(return_value=False)
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
result = await _init_flow(
hass,
source=SOURCE_IMPORT,
data={
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def test_options(hass: HomeAssistantType) -> None:
"""Check an options flow."""
config_entry = add_test_config_entry(hass)
client = create_mock_client()
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(TEST_ENTITY_ID_1) is not None
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
new_priority = 1
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={CONF_PRIORITY: new_priority}
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == {CONF_PRIORITY: new_priority}
# Turn the light on and ensure the new priority is used.
client.async_send_set_color = AsyncMock(return_value=True)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: TEST_ENTITY_ID_1},
blocking=True,
)
assert client.async_send_set_color.call_args[1][CONF_PRIORITY] == new_priority

473
tests/components/hyperion/test_light.py
View file

@ -1,82 +1,306 @@
"""Tests for the Hyperion integration."""
import logging
from types import MappingProxyType
from typing import Any, Optional
from hyperion import const
from homeassistant.components.hyperion import light as hyperion_light
from homeassistant import setup
from homeassistant.components.hyperion import (
get_hyperion_unique_id,
light as hyperion_light,
)
from homeassistant.components.hyperion.const import DOMAIN, TYPE_HYPERION_LIGHT
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_EFFECT,
ATTR_HS_COLOR,
DOMAIN,
DOMAIN as LIGHT_DOMAIN,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF, SERVICE_TURN_ON
from homeassistant.setup import async_setup_component
from homeassistant.helpers.entity_registry import async_get_registry
from homeassistant.helpers.typing import HomeAssistantType
from tests.async_mock import AsyncMock, Mock, call, patch
from . import (
TEST_CONFIG_ENTRY_OPTIONS,
TEST_ENTITY_ID_1,
TEST_ENTITY_ID_2,
TEST_ENTITY_ID_3,
TEST_HOST,
TEST_INSTANCE_1,
TEST_INSTANCE_2,
TEST_INSTANCE_3,
TEST_PORT,
TEST_PRIORITY,
TEST_SYSINFO_ID,
TEST_YAML_ENTITY_ID,
TEST_YAML_NAME,
add_test_config_entry,
create_mock_client,
setup_test_config_entry,
)
TEST_HOST = "test-hyperion-host"
TEST_PORT = const.DEFAULT_PORT
TEST_NAME = "test_hyperion_name"
TEST_PRIORITY = 128
TEST_ENTITY_ID = f"{DOMAIN}.{TEST_NAME}"
from tests.async_mock import AsyncMock, call, patch # type: ignore[attr-defined]
_LOGGER = logging.getLogger(__name__)
def create_mock_client():
"""Create a mock Hyperion client."""
mock_client = Mock()
mock_client.async_client_connect = AsyncMock(return_value=True)
mock_client.adjustment = None
mock_client.effects = None
mock_client.id = "%s:%i" % (TEST_HOST, TEST_PORT)
return mock_client
def call_registered_callback(client, key, *args, **kwargs):
def _call_registered_callback(
client: AsyncMock, key: str, *args: Any, **kwargs: Any
) -> None:
"""Call a Hyperion entity callback that was registered with the client."""
return client.set_callbacks.call_args[0][0][key](*args, **kwargs)
client.set_callbacks.call_args[0][0][key](*args, **kwargs)
async def setup_entity(hass, client=None):
async def _setup_entity_yaml(hass: HomeAssistantType, client: AsyncMock = None) -> None:
"""Add a test Hyperion entity to hass."""
client = client or create_mock_client()
with patch("hyperion.client.HyperionClient", return_value=client):
assert await async_setup_component(
with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client
):
assert await setup.async_setup_component(
hass,
DOMAIN,
LIGHT_DOMAIN,
{
DOMAIN: {
LIGHT_DOMAIN: {
"platform": "hyperion",
"name": TEST_NAME,
"name": TEST_YAML_NAME,
"host": TEST_HOST,
"port": const.DEFAULT_PORT,
"port": TEST_PORT,
"priority": TEST_PRIORITY,
}
},
)
await hass.async_block_till_done()
await hass.async_block_till_done()
async def test_setup_platform(hass):
"""Test setting up the platform."""
def _get_config_entry_from_unique_id(
hass: HomeAssistantType, unique_id: str
) -> Optional[ConfigEntry]:
for entry in hass.config_entries.async_entries(domain=DOMAIN):
if TEST_SYSINFO_ID == entry.unique_id:
return entry
return None
async def test_setup_yaml_already_converted(hass: HomeAssistantType) -> None:
"""Test an already converted YAML style config."""
# This tests "Possibility 1" from async_setup_platform()
# Add a pre-existing config entry.
add_test_config_entry(hass)
client = create_mock_client()
await setup_entity(hass, client=client)
assert hass.states.get(TEST_ENTITY_ID) is not None
await _setup_entity_yaml(hass, client=client)
# Setup should be skipped for the YAML config as there is a pre-existing config
# entry.
assert hass.states.get(TEST_YAML_ENTITY_ID) is None
async def test_setup_platform_not_ready(hass):
"""Test the platform not being ready."""
async def test_setup_yaml_old_style_unique_id(hass: HomeAssistantType) -> None:
"""Test an already converted YAML style config."""
# This tests "Possibility 2" from async_setup_platform()
old_unique_id = f"{TEST_HOST}:{TEST_PORT}-0"
# Add a pre-existing registry entry.
registry = await async_get_registry(hass)
registry.async_get_or_create(
domain=LIGHT_DOMAIN,
platform=DOMAIN,
unique_id=old_unique_id,
suggested_object_id=TEST_YAML_NAME,
)
client = create_mock_client()
await _setup_entity_yaml(hass, client=client)
# The entity should have been created with the same entity_id.
assert hass.states.get(TEST_YAML_ENTITY_ID) is not None
# The unique_id should have been updated in the registry (rather than the one
# specified above).
assert registry.async_get(TEST_YAML_ENTITY_ID).unique_id == get_hyperion_unique_id(
TEST_SYSINFO_ID, 0, TYPE_HYPERION_LIGHT
)
assert registry.async_get_entity_id(LIGHT_DOMAIN, DOMAIN, old_unique_id) is None
# There should be a config entry with the correct server unique_id.
entry = _get_config_entry_from_unique_id(hass, TEST_SYSINFO_ID)
assert entry
assert entry.options == MappingProxyType(TEST_CONFIG_ENTRY_OPTIONS)
async def test_setup_yaml_new_style_unique_id_wo_config(
hass: HomeAssistantType,
) -> None:
"""Test an a new unique_id without a config entry."""
# Note: This casde should not happen in the wild, as no released version of Home
# Assistant should this combination, but verify correct behavior for defense in
# depth.
new_unique_id = get_hyperion_unique_id(TEST_SYSINFO_ID, 0, TYPE_HYPERION_LIGHT)
entity_id_to_preserve = "light.magic_entity"
# Add a pre-existing registry entry.
registry = await async_get_registry(hass)
registry.async_get_or_create(
domain=LIGHT_DOMAIN,
platform=DOMAIN,
unique_id=new_unique_id,
suggested_object_id=entity_id_to_preserve.split(".")[1],
)
client = create_mock_client()
await _setup_entity_yaml(hass, client=client)
# The entity should have been created with the same entity_id.
assert hass.states.get(entity_id_to_preserve) is not None
# The unique_id should have been updated in the registry (rather than the one
# specified above).
assert registry.async_get(entity_id_to_preserve).unique_id == new_unique_id
# There should be a config entry with the correct server unique_id.
entry = _get_config_entry_from_unique_id(hass, TEST_SYSINFO_ID)
assert entry
assert entry.options == MappingProxyType(TEST_CONFIG_ENTRY_OPTIONS)
async def test_setup_yaml_no_registry_entity(hass: HomeAssistantType) -> None:
"""Test an already converted YAML style config."""
# This tests "Possibility 3" from async_setup_platform()
registry = await async_get_registry(hass)
# Add a pre-existing config entry.
client = create_mock_client()
await _setup_entity_yaml(hass, client=client)
# The entity should have been created with the same entity_id.
assert hass.states.get(TEST_YAML_ENTITY_ID) is not None
# The unique_id should have been updated in the registry (rather than the one
# specified above).
assert registry.async_get(TEST_YAML_ENTITY_ID).unique_id == get_hyperion_unique_id(
TEST_SYSINFO_ID, 0, TYPE_HYPERION_LIGHT
)
# There should be a config entry with the correct server unique_id.
entry = _get_config_entry_from_unique_id(hass, TEST_SYSINFO_ID)
assert entry
assert entry.options == MappingProxyType(TEST_CONFIG_ENTRY_OPTIONS)
async def test_setup_yaml_not_ready(hass: HomeAssistantType) -> None:
"""Test the component not being ready."""
client = create_mock_client()
client.async_client_connect = AsyncMock(return_value=False)
await setup_entity(hass, client=client)
assert hass.states.get(TEST_ENTITY_ID) is None
await _setup_entity_yaml(hass, client=client)
assert hass.states.get(TEST_YAML_ENTITY_ID) is None
async def test_light_basic_properies(hass):
async def test_setup_config_entry(hass: HomeAssistantType) -> None:
"""Test setting up the component via config entries."""
await setup_test_config_entry(hass, hyperion_client=create_mock_client())
assert hass.states.get(TEST_ENTITY_ID_1) is not None
async def test_setup_config_entry_not_ready(hass: HomeAssistantType) -> None:
"""Test the component not being ready."""
client = create_mock_client()
client.async_client_connect = AsyncMock(return_value=False)
await setup_test_config_entry(hass, hyperion_client=client)
assert hass.states.get(TEST_ENTITY_ID_1) is None
async def test_setup_config_entry_dynamic_instances(hass: HomeAssistantType) -> None:
"""Test dynamic changes in the omstamce configuration."""
config_entry = add_test_config_entry(hass)
master_client = create_mock_client()
master_client.instances = [TEST_INSTANCE_1, TEST_INSTANCE_2]
entity_client = create_mock_client()
entity_client.instances = master_client.instances
with patch(
"homeassistant.components.hyperion.client.HyperionClient",
side_effect=[master_client, entity_client, entity_client],
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(TEST_ENTITY_ID_1) is not None
assert hass.states.get(TEST_ENTITY_ID_2) is not None
# Inject a new instances update (remove instance 1, add instance 3)
assert master_client.set_callbacks.called
instance_callback = master_client.set_callbacks.call_args[0][0][
f"{const.KEY_INSTANCE}-{const.KEY_UPDATE}"
]
with patch(
"homeassistant.components.hyperion.client.HyperionClient",
return_value=entity_client,
):
instance_callback(
{
const.KEY_SUCCESS: True,
const.KEY_DATA: [TEST_INSTANCE_2, TEST_INSTANCE_3],
}
)
await hass.async_block_till_done()
assert hass.states.get(TEST_ENTITY_ID_1) is None
assert hass.states.get(TEST_ENTITY_ID_2) is not None
assert hass.states.get(TEST_ENTITY_ID_3) is not None
# Inject a new instances update (re-add instance 1, but not running)
with patch(
"homeassistant.components.hyperion.client.HyperionClient",
return_value=entity_client,
):
instance_callback(
{
const.KEY_SUCCESS: True,
const.KEY_DATA: [
{**TEST_INSTANCE_1, "running": False},
TEST_INSTANCE_2,
TEST_INSTANCE_3,
],
}
)
await hass.async_block_till_done()
assert hass.states.get(TEST_ENTITY_ID_1) is None
assert hass.states.get(TEST_ENTITY_ID_2) is not None
assert hass.states.get(TEST_ENTITY_ID_3) is not None
# Inject a new instances update (re-add instance 1, running)
with patch(
"homeassistant.components.hyperion.client.HyperionClient",
return_value=entity_client,
):
instance_callback(
{
const.KEY_SUCCESS: True,
const.KEY_DATA: [TEST_INSTANCE_1, TEST_INSTANCE_2, TEST_INSTANCE_3],
}
)
await hass.async_block_till_done()
assert hass.states.get(TEST_ENTITY_ID_1) is not None
assert hass.states.get(TEST_ENTITY_ID_2) is not None
assert hass.states.get(TEST_ENTITY_ID_3) is not None
async def test_light_basic_properies(hass: HomeAssistantType) -> None:
"""Test the basic properties."""
client = create_mock_client()
await setup_entity(hass, client=client)
await setup_test_config_entry(hass, hyperion_client=client)
entity_state = hass.states.get(TEST_ENTITY_ID)
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.state == "on"
assert entity_state.attributes["brightness"] == 255
assert entity_state.attributes["hs_color"] == (0.0, 0.0)
@ -91,15 +315,15 @@ async def test_light_basic_properies(hass):
)
async def test_light_async_turn_on(hass):
async def test_light_async_turn_on(hass: HomeAssistantType) -> None:
"""Test turning the light on."""
client = create_mock_client()
await setup_entity(hass, client=client)
await setup_test_config_entry(hass, hyperion_client=client)
# On (=), 100% (=), solid (=), [255,255,255] (=)
client.async_send_set_color = AsyncMock(return_value=True)
await hass.services.async_call(
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: TEST_ENTITY_ID}, blocking=True
LIGHT_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: TEST_ENTITY_ID_1}, blocking=True
)
assert client.async_send_set_color.call_args == call(
@ -116,9 +340,9 @@ async def test_light_async_turn_on(hass):
client.async_send_set_color = AsyncMock(return_value=True)
client.async_send_set_adjustment = AsyncMock(return_value=True)
await hass.services.async_call(
DOMAIN,
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: TEST_ENTITY_ID, ATTR_BRIGHTNESS: brightness},
{ATTR_ENTITY_ID: TEST_ENTITY_ID_1, ATTR_BRIGHTNESS: brightness},
blocking=True,
)
@ -135,8 +359,9 @@ async def test_light_async_turn_on(hass):
# Simulate a state callback from Hyperion.
client.adjustment = [{const.KEY_BRIGHTNESS: 50}]
call_registered_callback(client, "adjustment-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "adjustment-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.state == "on"
assert entity_state.attributes["brightness"] == brightness
@ -144,9 +369,9 @@ async def test_light_async_turn_on(hass):
hs_color = (180.0, 100.0)
client.async_send_set_color = AsyncMock(return_value=True)
await hass.services.async_call(
DOMAIN,
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: TEST_ENTITY_ID, ATTR_HS_COLOR: hs_color},
{ATTR_ENTITY_ID: TEST_ENTITY_ID_1, ATTR_HS_COLOR: hs_color},
blocking=True,
)
@ -164,8 +389,9 @@ async def test_light_async_turn_on(hass):
const.KEY_VALUE: {const.KEY_RGB: (0, 255, 255)},
}
call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["hs_color"] == hs_color
assert entity_state.attributes["icon"] == hyperion_light.ICON_LIGHTBULB
@ -175,9 +401,9 @@ async def test_light_async_turn_on(hass):
client.async_send_set_adjustment = AsyncMock(return_value=True)
await hass.services.async_call(
DOMAIN,
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: TEST_ENTITY_ID, ATTR_BRIGHTNESS: brightness},
{ATTR_ENTITY_ID: TEST_ENTITY_ID_1, ATTR_BRIGHTNESS: brightness},
blocking=True,
)
@ -192,8 +418,9 @@ async def test_light_async_turn_on(hass):
}
)
client.adjustment = [{const.KEY_BRIGHTNESS: 100}]
call_registered_callback(client, "adjustment-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "adjustment-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["brightness"] == brightness
# On (=), 100% (=), V4L (!), [0,255,255] (=)
@ -201,9 +428,9 @@ async def test_light_async_turn_on(hass):
client.async_send_clear = AsyncMock(return_value=True)
client.async_send_set_component = AsyncMock(return_value=True)
await hass.services.async_call(
DOMAIN,
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: TEST_ENTITY_ID, ATTR_EFFECT: effect},
{ATTR_ENTITY_ID: TEST_ENTITY_ID_1, ATTR_EFFECT: effect},
blocking=True,
)
@ -237,8 +464,9 @@ async def test_light_async_turn_on(hass):
),
]
client.visible_priority = {const.KEY_COMPONENTID: effect}
call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["icon"] == hyperion_light.ICON_EXTERNAL_SOURCE
assert entity_state.attributes["effect"] == effect
@ -248,9 +476,9 @@ async def test_light_async_turn_on(hass):
client.async_send_set_effect = AsyncMock(return_value=True)
await hass.services.async_call(
DOMAIN,
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: TEST_ENTITY_ID, ATTR_EFFECT: effect},
{ATTR_ENTITY_ID: TEST_ENTITY_ID_1, ATTR_EFFECT: effect},
blocking=True,
)
@ -268,33 +496,37 @@ async def test_light_async_turn_on(hass):
const.KEY_COMPONENTID: const.KEY_COMPONENTID_EFFECT,
const.KEY_OWNER: effect,
}
call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["icon"] == hyperion_light.ICON_EFFECT
assert entity_state.attributes["effect"] == effect
# No calls if disconnected.
client.has_loaded_state = False
call_registered_callback(client, "client-update", {"loaded-state": False})
_call_registered_callback(client, "client-update", {"loaded-state": False})
client.async_send_clear = AsyncMock(return_value=True)
client.async_send_set_effect = AsyncMock(return_value=True)
await hass.services.async_call(
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: TEST_ENTITY_ID}, blocking=True
LIGHT_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: TEST_ENTITY_ID_1}, blocking=True
)
assert not client.async_send_clear.called
assert not client.async_send_set_effect.called
async def test_light_async_turn_off(hass):
async def test_light_async_turn_off(hass: HomeAssistantType) -> None:
"""Test turning the light off."""
client = create_mock_client()
await setup_entity(hass, client=client)
await setup_test_config_entry(hass, hyperion_client=client)
client.async_send_set_component = AsyncMock(return_value=True)
await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: TEST_ENTITY_ID}, blocking=True
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: TEST_ENTITY_ID_1},
blocking=True,
)
assert client.async_send_set_component.call_args == call(
@ -309,50 +541,60 @@ async def test_light_async_turn_off(hass):
# No calls if no state loaded.
client.has_loaded_state = False
client.async_send_set_component = AsyncMock(return_value=True)
call_registered_callback(client, "client-update", {"loaded-state": False})
_call_registered_callback(client, "client-update", {"loaded-state": False})
await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: TEST_ENTITY_ID}, blocking=True
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: TEST_ENTITY_ID_1},
blocking=True,
)
assert not client.async_send_set_component.called
async def test_light_async_updates_from_hyperion_client(hass):
async def test_light_async_updates_from_hyperion_client(
hass: HomeAssistantType,
) -> None:
"""Test receiving a variety of Hyperion client callbacks."""
client = create_mock_client()
await setup_entity(hass, client=client)
await setup_test_config_entry(hass, hyperion_client=client)
# Bright change gets accepted.
brightness = 10
client.adjustment = [{const.KEY_BRIGHTNESS: brightness}]
call_registered_callback(client, "adjustment-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "adjustment-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["brightness"] == round(255 * (brightness / 100.0))
# Broken brightness value is ignored.
bad_brightness = -200
client.adjustment = [{const.KEY_BRIGHTNESS: bad_brightness}]
call_registered_callback(client, "adjustment-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "adjustment-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["brightness"] == round(255 * (brightness / 100.0))
# Update components.
client.is_on.return_value = True
call_registered_callback(client, "components-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "components-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.state == "on"
client.is_on.return_value = False
call_registered_callback(client, "components-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "components-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.state == "off"
# Update priorities (V4L)
client.is_on.return_value = True
client.visible_priority = {const.KEY_COMPONENTID: const.KEY_COMPONENTID_V4L}
call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["icon"] == hyperion_light.ICON_EXTERNAL_SOURCE
assert entity_state.attributes["hs_color"] == (0.0, 0.0)
assert entity_state.attributes["effect"] == const.KEY_COMPONENTID_V4L
@ -364,8 +606,9 @@ async def test_light_async_updates_from_hyperion_client(hass):
const.KEY_OWNER: effect,
}
call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["effect"] == effect
assert entity_state.attributes["icon"] == hyperion_light.ICON_EFFECT
assert entity_state.attributes["hs_color"] == (0.0, 0.0)
@ -377,8 +620,9 @@ async def test_light_async_updates_from_hyperion_client(hass):
const.KEY_VALUE: {const.KEY_RGB: rgb},
}
call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "priorities-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["effect"] == hyperion_light.KEY_EFFECT_SOLID
assert entity_state.attributes["icon"] == hyperion_light.ICON_LIGHTBULB
assert entity_state.attributes["hs_color"] == (180.0, 100.0)
@ -386,8 +630,9 @@ async def test_light_async_updates_from_hyperion_client(hass):
# Update effect list
effects = [{const.KEY_NAME: "One"}, {const.KEY_NAME: "Two"}]
client.effects = effects
call_registered_callback(client, "effects-update")
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "effects-update")
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["effect_list"] == [
effect[const.KEY_NAME] for effect in effects
] + const.KEY_COMPONENTID_EXTERNAL_SOURCES + [hyperion_light.KEY_EFFECT_SOLID]
@ -396,18 +641,20 @@ async def test_light_async_updates_from_hyperion_client(hass):
# Turn on late, check state, disconnect, ensure it cannot be turned off.
client.has_loaded_state = False
call_registered_callback(client, "client-update", {"loaded-state": False})
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "client-update", {"loaded-state": False})
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.state == "unavailable"
# Update connection status (e.g. re-connection)
client.has_loaded_state = True
call_registered_callback(client, "client-update", {"loaded-state": True})
entity_state = hass.states.get(TEST_ENTITY_ID)
_call_registered_callback(client, "client-update", {"loaded-state": True})
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.state == "on"
async def test_full_state_loaded_on_start(hass):
async def test_full_state_loaded_on_start(hass: HomeAssistantType) -> None:
"""Test receiving a variety of Hyperion client callbacks."""
client = create_mock_client()
@ -420,11 +667,43 @@ async def test_full_state_loaded_on_start(hass):
}
client.effects = [{const.KEY_NAME: "One"}, {const.KEY_NAME: "Two"}]
await setup_entity(hass, client=client)
entity_state = hass.states.get(TEST_ENTITY_ID)
await setup_test_config_entry(hass, hyperion_client=client)
entity_state = hass.states.get(TEST_ENTITY_ID_1)
assert entity_state
assert entity_state.attributes["brightness"] == round(255 * (brightness / 100.0))
assert entity_state.attributes["effect"] == hyperion_light.KEY_EFFECT_SOLID
assert entity_state.attributes["icon"] == hyperion_light.ICON_LIGHTBULB
assert entity_state.attributes["hs_color"] == (180.0, 100.0)
async def test_unload_entry(hass: HomeAssistantType) -> None:
"""Test unload."""
client = create_mock_client()
await setup_test_config_entry(hass, hyperion_client=client)
assert hass.states.get(TEST_ENTITY_ID_1) is not None
assert client.async_client_connect.called
assert not client.async_client_disconnect.called
entry = _get_config_entry_from_unique_id(hass, TEST_SYSINFO_ID)
assert entry
await hass.config_entries.async_unload(entry.entry_id)
assert client.async_client_disconnect.call_count == 2
async def test_version_log_warning(caplog, hass: HomeAssistantType) -> None:
"""Test warning on old version."""
client = create_mock_client()
client.async_sysinfo_version = AsyncMock(return_value="2.0.0-alpha.7")
await setup_test_config_entry(hass, hyperion_client=client)
assert hass.states.get(TEST_ENTITY_ID_1) is not None
assert "Please consider upgrading" in caplog.text
async def test_version_no_log_warning(caplog, hass: HomeAssistantType) -> None:
"""Test no warning on acceptable version."""
client = create_mock_client()
client.async_sysinfo_version = AsyncMock(return_value="2.0.0-alpha.9")
await setup_test_config_entry(hass, hyperion_client=client)
assert hass.states.get(TEST_ENTITY_ID_1) is not None
assert "Please consider upgrading" not in caplog.text