github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Convert Nanoleaf yaml and discovery to config flow (#52199) 

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Milan Meulemans 2021-08-24 21:09:36 +02:00 committed by GitHub
parent 9555a34691
commit 8570502681
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 761 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -685,7 +685,9 @@ omit =
homeassistant/components/myq/cover.py
homeassistant/components/myq/light.py
homeassistant/components/nad/media_player.py
homeassistant/components/nanoleaf/__init__.py
homeassistant/components/nanoleaf/light.py
homeassistant/components/nanoleaf/util.py
homeassistant/components/neato/__init__.py
homeassistant/components/neato/api.py
homeassistant/components/neato/camera.py

1
CODEOWNERS
View File

@ -326,6 +326,7 @@ homeassistant/components/myq/* @bdraco @ehendrix23
homeassistant/components/mysensors/* @MartinHjelmare @functionpointer
homeassistant/components/mystrom/* @fabaff
homeassistant/components/nam/* @bieniu
homeassistant/components/nanoleaf/* @milanmeu
homeassistant/components/neato/* @dshokouhi @Santobert
homeassistant/components/nederlandse_spoorwegen/* @YarmoM
homeassistant/components/nello/* @pschmitt

2
homeassistant/components/discovery/__init__.py
View File

@ -55,7 +55,6 @@ SERVICE_HANDLERS = {
"bose_soundtouch": ("media_player", "soundtouch"),
"bluesound": ("media_player", "bluesound"),
"lg_smart_device": ("media_player", "lg_soundbar"),
"nanoleaf_aurora": ("light", "nanoleaf"),
}
OPTIONAL_SERVICE_HANDLERS = {SERVICE_DLNA_DMR: ("media_player", "dlna_dmr")}
@ -87,6 +86,7 @@ MIGRATED_SERVICE_HANDLERS = [
SERVICE_XIAOMI_GW,
"volumio",
SERVICE_YEELIGHT,
"nanoleaf_aurora",
]
DEFAULT_ENABLED = (

32
homeassistant/components/nanoleaf/__init__.py
View File

@ -1 +1,31 @@
"""The nanoleaf component."""
"""The Nanoleaf integration."""
from pynanoleaf.pynanoleaf import Nanoleaf, Unavailable
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .const import DEVICE, DOMAIN, NAME, SERIAL_NO
from .util import pynanoleaf_get_info
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Nanoleaf from a config entry."""
nanoleaf = Nanoleaf(entry.data[CONF_HOST])
nanoleaf.token = entry.data[CONF_TOKEN]
try:
info = await hass.async_add_executor_job(pynanoleaf_get_info, nanoleaf)
except Unavailable as err:
raise ConfigEntryNotReady from err
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
DEVICE: nanoleaf,
NAME: info["name"],
SERIAL_NO: info["serialNo"],
}
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "light")
)
return True

204
homeassistant/components/nanoleaf/config_flow.py Normal file
View File

@ -0,0 +1,204 @@
"""Config flow for Nanoleaf integration."""
from __future__ import annotations
import logging
import os
from typing import Any, Final, cast
from pynanoleaf import InvalidToken, Nanoleaf, NotAuthorizingNewTokens, Unavailable
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import persistent_notification
from homeassistant.const import CONF_HOST, CONF_TOKEN
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.typing import DiscoveryInfoType
from homeassistant.util.json import load_json, save_json
from .const import DOMAIN
from .util import pynanoleaf_get_info
_LOGGER = logging.getLogger(__name__)
# For discovery integration import
CONFIG_FILE: Final = ".nanoleaf.conf"
USER_SCHEMA: Final = vol.Schema(
{
vol.Required(CONF_HOST): str,
}
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Nanoleaf config flow."""
VERSION = 1
def __init__(self) -> None:
"""Initialize a Nanoleaf flow."""
self.nanoleaf: Nanoleaf
# For discovery integration import
self.discovery_conf: dict
self.device_id: str
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle Nanoleaf flow initiated by the user."""
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=USER_SCHEMA, last_step=False
)
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
self.nanoleaf = Nanoleaf(user_input[CONF_HOST])
try:
await self.hass.async_add_executor_job(self.nanoleaf.authorize)
except Unavailable:
return self.async_show_form(
step_id="user",
data_schema=USER_SCHEMA,
errors={"base": "cannot_connect"},
last_step=False,
)
except NotAuthorizingNewTokens:
pass
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unknown error connecting to Nanoleaf")
return self.async_show_form(
step_id="user",
data_schema=USER_SCHEMA,
last_step=False,
errors={"base": "unknown"},
)
return await self.async_step_link()
async def async_step_zeroconf(
self, discovery_info: DiscoveryInfoType
) -> FlowResult:
"""Handle Nanoleaf Zeroconf discovery."""
_LOGGER.debug("Zeroconf discovered: %s", discovery_info)
return await self._async_discovery_handler(discovery_info)
async def async_step_homekit(self, discovery_info: DiscoveryInfoType) -> FlowResult:
"""Handle Nanoleaf Homekit discovery."""
_LOGGER.debug("Homekit discovered: %s", discovery_info)
return await self._async_discovery_handler(discovery_info)
async def _async_discovery_handler(
self, discovery_info: DiscoveryInfoType
) -> FlowResult:
"""Handle Nanoleaf discovery."""
host = discovery_info["host"]
# The name is unique and printed on the device and cannot be changed.
name = discovery_info["name"].replace(f".{discovery_info['type']}", "")
await self.async_set_unique_id(name)
self._abort_if_unique_id_configured({CONF_HOST: host})
self.nanoleaf = Nanoleaf(host)
# Import from discovery integration
self.device_id = discovery_info["properties"]["id"]
self.discovery_conf = cast(
dict,
await self.hass.async_add_executor_job(
load_json, self.hass.config.path(CONFIG_FILE)
),
)
self.nanoleaf.token = self.discovery_conf.get(self.device_id, {}).get(
"token", # >= 2021.4
self.discovery_conf.get(host, {}).get("token"), # < 2021.4
)
if self.nanoleaf.token is not None:
_LOGGER.warning(
"Importing Nanoleaf %s from the discovery integration", name
)
return await self.async_setup_finish(discovery_integration_import=True)
self.context["title_placeholders"] = {"name": name}
return await self.async_step_link()
async def async_step_link(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle Nanoleaf link step."""
if user_input is None:
return self.async_show_form(step_id="link")
try:
await self.hass.async_add_executor_job(self.nanoleaf.authorize)
except NotAuthorizingNewTokens:
return self.async_show_form(
step_id="link", errors={"base": "not_allowing_new_tokens"}
)
except Unavailable:
return self.async_abort(reason="cannot_connect")
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unknown error authorizing Nanoleaf")
return self.async_show_form(step_id="link", errors={"base": "unknown"})
return await self.async_setup_finish()
async def async_step_import(self, config: dict[str, Any]) -> FlowResult:
"""Handle Nanoleaf configuration import."""
self._async_abort_entries_match({CONF_HOST: config[CONF_HOST]})
_LOGGER.debug(
"Importing Nanoleaf on %s from your configuration.yaml", config[CONF_HOST]
)
self.nanoleaf = Nanoleaf(config[CONF_HOST])
self.nanoleaf.token = config[CONF_TOKEN]
return await self.async_setup_finish()
async def async_setup_finish(
self, discovery_integration_import: bool = False
) -> FlowResult:
"""Finish Nanoleaf config flow."""
try:
info = await self.hass.async_add_executor_job(
pynanoleaf_get_info, self.nanoleaf
)
except Unavailable:
return self.async_abort(reason="cannot_connect")
except InvalidToken:
return self.async_abort(reason="invalid_token")
except Exception: # pylint: disable=broad-except
_LOGGER.exception(
"Unknown error connecting with Nanoleaf at %s with token %s",
self.nanoleaf.host,
self.nanoleaf.token,
)
return self.async_abort(reason="unknown")
name = info["name"]
await self.async_set_unique_id(name)
self._abort_if_unique_id_configured({CONF_HOST: self.nanoleaf.host})
if discovery_integration_import:
if self.nanoleaf.host in self.discovery_conf:
self.discovery_conf.pop(self.nanoleaf.host)
if self.device_id in self.discovery_conf:
self.discovery_conf.pop(self.device_id)
_LOGGER.info(
"Successfully imported Nanoleaf %s from the discovery integration",
name,
)
if self.discovery_conf:
await self.hass.async_add_executor_job(
save_json, self.hass.config.path(CONFIG_FILE), self.discovery_conf
)
else:
await self.hass.async_add_executor_job(
os.remove, self.hass.config.path(CONFIG_FILE)
)
persistent_notification.async_create(
self.hass,
"All Nanoleaf devices from the discovery integration are imported. If you used the discovery integration only for Nanoleaf you can remove it from your configuration.yaml",
f"Imported Nanoleaf {name}",
)
return self.async_create_entry(
title=name,
data={
CONF_HOST: self.nanoleaf.host,
CONF_TOKEN: self.nanoleaf.token,
},
)

7
homeassistant/components/nanoleaf/const.py Normal file
View File

@ -0,0 +1,7 @@
"""Constants for Nanoleaf integration."""
DOMAIN = "nanoleaf"
DEVICE = "device"
SERIAL_NO = "serial_no"
NAME = "name"

93
homeassistant/components/nanoleaf/light.py
View File

@ -1,7 +1,9 @@
"""Support for Nanoleaf Lights."""
from __future__ import annotations
import logging
from pynanoleaf import Nanoleaf, Unavailable
from pynanoleaf import Unavailable
import voluptuous as vol
from homeassistant.components.light import (
@ -18,22 +20,23 @@ from homeassistant.components.light import (
SUPPORT_TRANSITION,
LightEntity,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_TOKEN
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import color as color_util
from homeassistant.util.color import (
color_temperature_mired_to_kelvin as mired_to_kelvin,
)
from homeassistant.util.json import load_json, save_json
from .const import DEVICE, DOMAIN, NAME, SERIAL_NO
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "Nanoleaf"
DATA_NANOLEAF = "nanoleaf"
CONFIG_FILE = ".nanoleaf.conf"
ICON = "mdi:triangle-outline"
SUPPORT_NANOLEAF = (
@ -53,69 +56,34 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Import Nanoleaf light platform."""
await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={CONF_HOST: config[CONF_HOST], CONF_TOKEN: config[CONF_TOKEN]},
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, add_entities: AddEntitiesCallback
) -> None:
"""Set up the Nanoleaf light."""
if DATA_NANOLEAF not in hass.data:
hass.data[DATA_NANOLEAF] = {}
token = ""
if discovery_info is not None:
host = discovery_info["host"]
name = None
device_id = discovery_info["properties"]["id"]
# if device already exists via config, skip discovery setup
if host in hass.data[DATA_NANOLEAF]:
return
_LOGGER.info("Discovered a new Nanoleaf: %s", discovery_info)
conf = load_json(hass.config.path(CONFIG_FILE))
if host in conf and device_id not in conf:
conf[device_id] = conf.pop(host)
save_json(hass.config.path(CONFIG_FILE), conf)
token = conf.get(device_id, {}).get("token", "")
else:
host = config[CONF_HOST]
name = config[CONF_NAME]
token = config[CONF_TOKEN]
nanoleaf_light = Nanoleaf(host)
if not token:
token = nanoleaf_light.request_token()
if not token:
_LOGGER.error(
"Could not generate the auth token, did you press "
"and hold the power button on %s"
"for 5-7 seconds?",
name,
)
return
conf = load_json(hass.config.path(CONFIG_FILE))
conf[host] = {"token": token}
save_json(hass.config.path(CONFIG_FILE), conf)
nanoleaf_light.token = token
try:
info = nanoleaf_light.info
except Unavailable:
_LOGGER.error("Could not connect to Nanoleaf Light: %s on %s", name, host)
return
if name is None:
name = info.name
hass.data[DATA_NANOLEAF][host] = nanoleaf_light
add_entities([NanoleafLight(nanoleaf_light, name)], True)
data = hass.data[DOMAIN][entry.entry_id]
add_entities([NanoleafLight(data[DEVICE], data[NAME], data[SERIAL_NO])], True)
class NanoleafLight(LightEntity):
"""Representation of a Nanoleaf Light."""
def __init__(self, light, name):
def __init__(self, light, name, unique_id):
"""Initialize an Nanoleaf light."""
self._unique_id = light.serialNo
self._unique_id = unique_id
self._available = True
self._brightness = None
self._color_temp = None
@ -239,7 +207,6 @@ class NanoleafLight(LightEntity):
def update(self):
"""Fetch new state data for this light."""
try:
self._available = self._light.available
self._brightness = self._light.brightness

11
homeassistant/components/nanoleaf/manifest.json
View File

@ -1,8 +1,15 @@
{
"domain": "nanoleaf",
"name": "Nanoleaf",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/nanoleaf",
"requirements": ["pynanoleaf==0.1.0"],
"codeowners": [],
"zeroconf": ["_nanoleafms._tcp.local.", "_nanoleafapi._tcp.local."],
"homekit" : {
"models": [
"NL*"
]
},
"codeowners": ["@milanmeu"],
"iot_class": "local_polling"
}
}

27
homeassistant/components/nanoleaf/strings.json Normal file
View File

@ -0,0 +1,27 @@
{
"config": {
"flow_title": "{name}",
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]"
}
},
"link": {
"title": "Link Nanoleaf",
"description": "Press and hold the power button on your Nanoleaf for 5 seconds until the button LEDs start flashing, then click **SUBMIT** within 30 seconds."
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"not_allowing_new_tokens": "Nanoleaf is not allowing new tokens, follow the instructions above.",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_token": "[%key:common::config_flow::error::invalid_access_token%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
}
}

27
homeassistant/components/nanoleaf/translations/en.json Normal file
View File

@ -0,0 +1,27 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"cannot_connect": "Failed to connect",
"invalid_token": "Invalid access token",
"unknown": "Unexpected error"
},
"error": {
"cannot_connect": "Failed to connect",
"not_allowing_new_tokens": "Nanoleaf is not allowing new tokens, follow the instructions above.",
"unknown": "Unexpected error"
},
"flow_title": "{name}",
"step": {
"link": {
"description": "Press and hold the power button on your Nanoleaf for 5 seconds until the button LEDs start flashing, then click **SUBMIT** within 30 seconds.",
"title": "Link Nanoleaf"
},
"user": {
"data": {
"host": "Host"
}
}
}
}
}

7
homeassistant/components/nanoleaf/util.py Normal file
View File

@ -0,0 +1,7 @@
"""Nanoleaf integration util."""
from pynanoleaf.pynanoleaf import Nanoleaf
def pynanoleaf_get_info(nanoleaf_light: Nanoleaf) -> dict:
"""Get Nanoleaf light info."""
return nanoleaf_light.info

1
homeassistant/generated/config_flows.py
View File

@ -176,6 +176,7 @@ FLOWS = [
"myq",
"mysensors",
"nam",
"nanoleaf",
"neato",
"nest",
"netatmo",

11
homeassistant/generated/zeroconf.py
View File

@ -165,6 +165,16 @@ ZEROCONF = {
"domain": "xiaomi_miio"
}
],
"_nanoleafapi._tcp.local.": [
{
"domain": "nanoleaf"
}
],
"_nanoleafms._tcp.local.": [
{
"domain": "nanoleaf"
}
],
"_nut._tcp.local.": [
{
"domain": "nut"
@ -251,6 +261,7 @@ HOMEKIT = {
"Iota": "abode",
"LIFX": "lifx",
"MYQ": "myq",
"NL*": "nanoleaf",
"Netatmo Relay": "netatmo",
"PowerView": "hunterdouglas_powerview",
"Presence": "netatmo",

3
requirements_test_all.txt
View File

@ -941,6 +941,9 @@ pymyq==3.1.3
# homeassistant.components.mysensors
pymysensors==0.21.0
# homeassistant.components.nanoleaf
pynanoleaf==0.1.0
# homeassistant.components.nuki
pynuki==1.4.1

1
tests/components/nanoleaf/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Nanoleaf integration."""

399
tests/components/nanoleaf/test_config_flow.py Normal file
View File

@ -0,0 +1,399 @@
"""Test the Nanoleaf config flow."""
from unittest.mock import patch
from pynanoleaf import InvalidToken, NotAuthorizingNewTokens, Unavailable
from homeassistant import config_entries
from homeassistant.components.nanoleaf.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_TOKEN
from homeassistant.core import HomeAssistant
TEST_NAME = "Canvas ADF9"
TEST_HOST = "192.168.0.100"
TEST_TOKEN = "R34F1c92FNv3pcZs4di17RxGqiLSwHM"
TEST_OTHER_TOKEN = "Qs4dxGcHR34l29RF1c92FgiLQBt3pcM"
TEST_DEVICE_ID = "5E:2E:EA:XX:XX:XX"
TEST_OTHER_DEVICE_ID = "5E:2E:EA:YY:YY:YY"
async def test_user_unavailable_user_step(hass: HomeAssistant) -> None:
"""Test we handle Unavailable errors when host is not available in user step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
side_effect=Unavailable("message"),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: TEST_HOST,
},
)
assert result2["type"] == "form"
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "cannot_connect"}
assert not result2["last_step"]
async def test_user_unavailable_link_step(hass: HomeAssistant) -> None:
"""Test we abort if the device becomes unavailable in the link step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
return_value=None,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: TEST_HOST,
},
)
assert result2["type"] == "form"
assert result2["step_id"] == "link"
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
side_effect=Unavailable("message"),
):
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
assert result3["type"] == "abort"
assert result3["reason"] == "cannot_connect"
async def test_user_unavailable_setup_finish(hass: HomeAssistant) -> None:
"""Test we abort if the device becomes unavailable during setup_finish."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
return_value=None,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: TEST_HOST,
},
)
assert result2["type"] == "form"
assert result2["step_id"] == "link"
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
return_value=None,
), patch(
"homeassistant.components.nanoleaf.config_flow.pynanoleaf_get_info",
side_effect=Unavailable("message"),
):
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
assert result3["type"] == "abort"
assert result3["reason"] == "cannot_connect"
async def test_user_not_authorizing_new_tokens(hass: HomeAssistant) -> None:
"""Test we handle NotAuthorizingNewTokens errors."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] is None
assert not result["last_step"]
assert result["step_id"] == "user"
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
side_effect=NotAuthorizingNewTokens("message"),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: TEST_HOST,
},
)
assert result2["type"] == "form"
assert result2["errors"] is None
assert result2["step_id"] == "link"
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
)
assert result3["type"] == "form"
assert result3["errors"] is None
assert result3["step_id"] == "link"
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
side_effect=NotAuthorizingNewTokens("message"),
):
result4 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
assert result4["type"] == "form"
assert result4["step_id"] == "link"
assert result4["errors"] == {"base": "not_allowing_new_tokens"}
async def test_user_exception(hass: HomeAssistant) -> None:
"""Test we handle Exception errors."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
side_effect=Exception,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: TEST_HOST,
},
)
assert result2["type"] == "form"
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "unknown"}
assert not result2["last_step"]
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
return_value=None,
):
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: TEST_HOST,
},
)
assert result3["step_id"] == "link"
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
side_effect=Exception,
):
result4 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
assert result4["type"] == "form"
assert result4["step_id"] == "link"
assert result4["errors"] == {"base": "unknown"}
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
return_value=None,
), patch(
"homeassistant.components.nanoleaf.config_flow.pynanoleaf_get_info",
side_effect=Exception,
):
result5 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
assert result5["type"] == "abort"
assert result5["reason"] == "unknown"
async def test_zeroconf_discovery(hass: HomeAssistant) -> None:
"""Test zeroconfig discovery flow init."""
zeroconf = "_nanoleafms._tcp.local"
with patch(
"homeassistant.components.nanoleaf.config_flow.pynanoleaf_get_info",
return_value={"name": TEST_NAME},
), patch(
"homeassistant.components.nanoleaf.config_flow.load_json",
return_value={},
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data={
"host": TEST_HOST,
"name": f"{TEST_NAME}.{zeroconf}",
"type": zeroconf,
"properties": {"id": TEST_DEVICE_ID},
},
)
assert result["type"] == "form"
assert result["step_id"] == "link"
async def test_homekit_discovery_link_unavailable(
hass: HomeAssistant,
) -> None:
"""Test homekit discovery and abort if device is unavailable."""
homekit = "_hap._tcp.local"
with patch(
"homeassistant.components.nanoleaf.config_flow.pynanoleaf_get_info",
return_value={"name": TEST_NAME},
), patch(
"homeassistant.components.nanoleaf.config_flow.load_json",
return_value={},
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_HOMEKIT},
data={
"host": TEST_HOST,
"name": f"{TEST_NAME}.{homekit}",
"type": homekit,
"properties": {"id": TEST_DEVICE_ID},
},
)
assert result["type"] == "form"
assert result["step_id"] == "link"
context = next(
flow["context"]
for flow in hass.config_entries.flow.async_progress()
if flow["flow_id"] == result["flow_id"]
)
assert context["title_placeholders"] == {"name": TEST_NAME}
assert context["unique_id"] == TEST_NAME
with patch(
"homeassistant.components.nanoleaf.config_flow.Nanoleaf.authorize",
side_effect=Unavailable("message"),
):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == "abort"
assert result["reason"] == "cannot_connect"
async def test_import_config(hass: HomeAssistant) -> None:
"""Test configuration import."""
with patch(
"homeassistant.components.nanoleaf.config_flow.pynanoleaf_get_info",
return_value={"name": TEST_NAME},
), patch(
"homeassistant.components.nanoleaf.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={CONF_HOST: TEST_HOST, CONF_TOKEN: TEST_TOKEN},
)
assert result["type"] == "create_entry"
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_TOKEN: TEST_TOKEN,
}
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_config_invalid_token(hass: HomeAssistant) -> None:
"""Test configuration import with invalid token."""
with patch(
"homeassistant.components.nanoleaf.config_flow.pynanoleaf_get_info",
side_effect=InvalidToken("message"),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={CONF_HOST: TEST_HOST, CONF_TOKEN: TEST_TOKEN},
)
assert result["type"] == "abort"
assert result["reason"] == "invalid_token"
async def test_import_last_discovery_integration_host_zeroconf(
hass: HomeAssistant,
) -> None:
"""
Test discovery integration import from < 2021.4 (host) with zeroconf.
Device is last in Nanoleaf config file.
"""
zeroconf = "_nanoleafapi._tcp.local"
with patch(
"homeassistant.components.nanoleaf.config_flow.load_json",
return_value={TEST_HOST: {"token": TEST_TOKEN}},
), patch(
"homeassistant.components.nanoleaf.config_flow.pynanoleaf_get_info",
return_value={"name": TEST_NAME},
), patch(
"homeassistant.components.nanoleaf.config_flow.os.remove",
return_value=None,
) as mock_remove, patch(
"homeassistant.components.nanoleaf.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data={
"host": TEST_HOST,
"name": f"{TEST_NAME}.{zeroconf}",
"type": zeroconf,
"properties": {"id": TEST_DEVICE_ID},
},
)
assert result["type"] == "create_entry"
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_TOKEN: TEST_TOKEN,
}
mock_remove.assert_called_once()
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_not_last_discovery_integration_device_id_homekit(
hass: HomeAssistant,
) -> None:
"""
Test discovery integration import from >= 2021.4 (device_id) with homekit.
Device is not the only one in the Nanoleaf config file.
"""
homekit = "_hap._tcp.local"
with patch(
"homeassistant.components.nanoleaf.config_flow.load_json",
return_value={
TEST_DEVICE_ID: {"token": TEST_TOKEN},
TEST_OTHER_DEVICE_ID: {"token": TEST_OTHER_TOKEN},
},
), patch(
"homeassistant.components.nanoleaf.config_flow.pynanoleaf_get_info",
return_value={"name": TEST_NAME},
), patch(
"homeassistant.components.nanoleaf.config_flow.save_json",
return_value=None,
) as mock_save_json, patch(
"homeassistant.components.nanoleaf.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_HOMEKIT},
data={
"host": TEST_HOST,
"name": f"{TEST_NAME}.{homekit}",
"type": homekit,
"properties": {"id": TEST_DEVICE_ID},
},
)
assert result["type"] == "create_entry"
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_TOKEN: TEST_TOKEN,
}
mock_save_json.assert_called_once()
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1