feat(config_flow): Einrichtung per IP und mDNS, Neuanmeldung, Neukonfiguration, Optionen
This commit is contained in:
Michael Sauer
@@ -1,13 +1,204 @@
|
||||
"""Minimal config flow stub; Task 5 replaces this with the real flow."""
|
||||
"""Config flow for ninlight lamps."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.config_entries import ConfigFlow
|
||||
from collections.abc import Mapping
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from .const import DOMAIN
|
||||
from homeassistant.config_entries import (
|
||||
ConfigEntry,
|
||||
ConfigFlow,
|
||||
ConfigFlowResult,
|
||||
OptionsFlowWithReload,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_TOKEN
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.device_registry import format_mac
|
||||
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
|
||||
import voluptuous as vol
|
||||
|
||||
from .api import NinlightAuthError, NinlightClient, NinlightError, NinlightResponseError
|
||||
from .const import (
|
||||
CONF_LAMP_BASE_URL,
|
||||
CONF_SCAN_INTERVAL,
|
||||
DEFAULT_SCAN_INTERVAL,
|
||||
DEFAULT_TOKEN,
|
||||
DOMAIN,
|
||||
MAX_SCAN_INTERVAL,
|
||||
MIN_SCAN_INTERVAL,
|
||||
)
|
||||
from .entity import device_name
|
||||
from .urls import validate_lamp_base_url
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class NinlightConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Placeholder so manifest.json's config_flow: true can be imported."""
|
||||
"""One config entry per lamp, identified by its MAC."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._host: str = ""
|
||||
self._discovered_name: str = ""
|
||||
|
||||
@staticmethod
|
||||
@callback
|
||||
def async_get_options_flow(config_entry: ConfigEntry) -> NinlightOptionsFlow:
|
||||
return NinlightOptionsFlow()
|
||||
|
||||
async def _async_probe(
|
||||
self, host: str, token: str
|
||||
) -> tuple[dict[str, Any] | None, dict[str, str]]:
|
||||
"""Check the token (system/settings always enforces it), then read the identity."""
|
||||
client = NinlightClient(async_get_clientsession(self.hass), host, token)
|
||||
try:
|
||||
await client.get_settings()
|
||||
info = await client.get_info()
|
||||
except NinlightAuthError:
|
||||
return None, {"base": "invalid_auth"}
|
||||
except NinlightResponseError:
|
||||
return None, {"base": "not_ninlight"}
|
||||
except NinlightError:
|
||||
return None, {"base": "cannot_connect"}
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected error while probing %s", host)
|
||||
return None, {"base": "unknown"}
|
||||
return info, {}
|
||||
|
||||
async def async_step_user(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
host = user_input[CONF_HOST].strip()
|
||||
token = user_input[CONF_TOKEN]
|
||||
info, errors = await self._async_probe(host, token)
|
||||
if info is not None:
|
||||
await self.async_set_unique_id(format_mac(info["mac"]))
|
||||
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
|
||||
return self.async_create_entry(
|
||||
title=device_name(info), data={CONF_HOST: host, CONF_TOKEN: token}
|
||||
)
|
||||
defaults = user_input or {}
|
||||
schema = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST, default=defaults.get(CONF_HOST, "")): str,
|
||||
vol.Required(CONF_TOKEN, default=defaults.get(CONF_TOKEN, DEFAULT_TOKEN)): str,
|
||||
}
|
||||
)
|
||||
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
|
||||
|
||||
async def async_step_zeroconf(self, discovery_info: ZeroconfServiceInfo) -> ConfigFlowResult:
|
||||
properties = discovery_info.properties
|
||||
mac = properties.get("mac")
|
||||
if not mac:
|
||||
return self.async_abort(reason="not_ninlight")
|
||||
host = str(discovery_info.ip_address)
|
||||
await self.async_set_unique_id(format_mac(mac))
|
||||
# Known lamp under a new address: update the entry silently and stop here.
|
||||
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
|
||||
self._host = host
|
||||
self._discovered_name = (
|
||||
properties.get("name") or properties.get("type") or discovery_info.name
|
||||
)
|
||||
self.context["title_placeholders"] = {"name": self._discovered_name}
|
||||
return await self.async_step_zeroconf_confirm()
|
||||
|
||||
async def async_step_zeroconf_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
info, errors = await self._async_probe(self._host, user_input[CONF_TOKEN])
|
||||
if info is not None:
|
||||
await self.async_set_unique_id(format_mac(info["mac"]), raise_on_progress=False)
|
||||
self._abort_if_unique_id_configured(updates={CONF_HOST: self._host})
|
||||
return self.async_create_entry(
|
||||
title=device_name(info),
|
||||
data={CONF_HOST: self._host, CONF_TOKEN: user_input[CONF_TOKEN]},
|
||||
)
|
||||
return self.async_show_form(
|
||||
step_id="zeroconf_confirm",
|
||||
data_schema=vol.Schema({vol.Required(CONF_TOKEN, default=DEFAULT_TOKEN): str}),
|
||||
description_placeholders={"name": self._discovered_name, "host": self._host},
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> ConfigFlowResult:
|
||||
return await self.async_step_reauth_confirm()
|
||||
|
||||
async def async_step_reauth_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
entry = self._get_reauth_entry()
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
info, errors = await self._async_probe(entry.data[CONF_HOST], user_input[CONF_TOKEN])
|
||||
if info is not None:
|
||||
await self.async_set_unique_id(format_mac(info["mac"]))
|
||||
self._abort_if_unique_id_mismatch(reason="wrong_device")
|
||||
return self.async_update_reload_and_abort(
|
||||
entry, data_updates={CONF_TOKEN: user_input[CONF_TOKEN]}
|
||||
)
|
||||
return self.async_show_form(
|
||||
step_id="reauth_confirm",
|
||||
data_schema=vol.Schema({vol.Required(CONF_TOKEN): str}),
|
||||
description_placeholders={"name": entry.title},
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_reconfigure(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
entry = self._get_reconfigure_entry()
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
host = user_input[CONF_HOST].strip()
|
||||
info, errors = await self._async_probe(host, user_input[CONF_TOKEN])
|
||||
if info is not None:
|
||||
await self.async_set_unique_id(format_mac(info["mac"]))
|
||||
self._abort_if_unique_id_mismatch(reason="wrong_device")
|
||||
return self.async_update_reload_and_abort(
|
||||
entry, data_updates={CONF_HOST: host, CONF_TOKEN: user_input[CONF_TOKEN]}
|
||||
)
|
||||
schema = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST, default=entry.data[CONF_HOST]): str,
|
||||
vol.Required(CONF_TOKEN, default=entry.data[CONF_TOKEN]): str,
|
||||
}
|
||||
)
|
||||
return self.async_show_form(step_id="reconfigure", data_schema=schema, errors=errors)
|
||||
|
||||
|
||||
class NinlightOptionsFlow(OptionsFlowWithReload):
|
||||
"""Polling interval and the address lamps use to reach Home Assistant."""
|
||||
|
||||
async def async_step_init(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
base = str(user_input.get(CONF_LAMP_BASE_URL) or "").strip()
|
||||
error = validate_lamp_base_url(base) if base else None
|
||||
if error:
|
||||
errors[CONF_LAMP_BASE_URL] = error
|
||||
else:
|
||||
data: dict[str, Any] = {CONF_SCAN_INTERVAL: user_input[CONF_SCAN_INTERVAL]}
|
||||
if base:
|
||||
data[CONF_LAMP_BASE_URL] = base.rstrip("/")
|
||||
return self.async_create_entry(data=data)
|
||||
current = self.config_entry.options
|
||||
schema = vol.Schema(
|
||||
{
|
||||
vol.Required(
|
||||
CONF_SCAN_INTERVAL,
|
||||
default=current.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL),
|
||||
): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=MIN_SCAN_INTERVAL, max=MAX_SCAN_INTERVAL)
|
||||
),
|
||||
vol.Optional(
|
||||
CONF_LAMP_BASE_URL,
|
||||
description={"suggested_value": current.get(CONF_LAMP_BASE_URL, "")},
|
||||
): str,
|
||||
}
|
||||
)
|
||||
return self.async_show_form(step_id="init", data_schema=schema, errors=errors)
|
||||
|
||||
+11
-1
@@ -23,7 +23,10 @@ from tests_unit.samples import (
|
||||
TOKEN,
|
||||
)
|
||||
|
||||
CLIENT_PATCH_TARGETS = ["custom_components.ninlight.NinlightClient"]
|
||||
CLIENT_PATCH_TARGETS = [
|
||||
"custom_components.ninlight.NinlightClient",
|
||||
"custom_components.ninlight.config_flow.NinlightClient",
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
@@ -91,3 +94,10 @@ async def init_integration(
|
||||
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
return mock_config_entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock]:
|
||||
"""Flow tests must not set the entry up for real."""
|
||||
with patch("custom_components.ninlight.async_setup_entry", return_value=True) as mock:
|
||||
yield mock
|
||||
|
||||
@@ -0,0 +1,298 @@
|
||||
"""Config flow: manual, zeroconf, reauth, reconfigure and options."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from ipaddress import ip_address
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
|
||||
from homeassistant.const import CONF_HOST, CONF_TOKEN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType, InvalidData
|
||||
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
|
||||
import pytest
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from custom_components.ninlight.api import (
|
||||
NinlightAuthError,
|
||||
NinlightConnectionError,
|
||||
NinlightResponseError,
|
||||
)
|
||||
from custom_components.ninlight.const import (
|
||||
CONF_LAMP_BASE_URL,
|
||||
CONF_SCAN_INTERVAL,
|
||||
DEFAULT_TOKEN,
|
||||
DOMAIN,
|
||||
)
|
||||
from tests_unit.samples import HOST_RGB, INFO_CW, MAC_RGB, TOKEN
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
|
||||
|
||||
NEW_TOKEN = "b71e0c9d44a2f6e3"
|
||||
|
||||
|
||||
def zeroconf_info(ip: str = HOST_RGB, **properties: Any) -> ZeroconfServiceInfo:
|
||||
props = {"type": "gu10_rgb", "fw": "0.3.8-36-gca9bb", "mac": MAC_RGB, "name": ""} | properties
|
||||
return ZeroconfServiceInfo(
|
||||
ip_address=ip_address(ip),
|
||||
ip_addresses=[ip_address(ip)],
|
||||
port=3333,
|
||||
hostname="ninlight-06b8.local.",
|
||||
type="_ninlight._tcp.local.",
|
||||
name="ninlight-06b8._ninlight._tcp.local.",
|
||||
properties={k: v for k, v in props.items() if v is not None},
|
||||
)
|
||||
|
||||
|
||||
def existing_entry(hass: HomeAssistant, host: str = HOST_RGB) -> MockConfigEntry:
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN, title="Lampe", unique_id=MAC_RGB, data={CONF_HOST: host, CONF_TOKEN: TOKEN}
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
return entry
|
||||
|
||||
|
||||
async def test_user_flow_creates_entry(hass: HomeAssistant, mock_client: MagicMock) -> None:
|
||||
result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_HOST: " 192.168.2.22 ", CONF_TOKEN: TOKEN}
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "Klyqa GU10 RGB CW/WW 06b8"
|
||||
assert result["data"] == {CONF_HOST: HOST_RGB, CONF_TOKEN: TOKEN}
|
||||
assert result["result"].unique_id == MAC_RGB
|
||||
mock_client.get_settings.assert_awaited_once() # the read that enforces the token
|
||||
|
||||
|
||||
async def test_user_form_prefills_default_token(
|
||||
hass: HomeAssistant, mock_client: MagicMock
|
||||
) -> None:
|
||||
result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
|
||||
schema = result["data_schema"].schema
|
||||
token_key = next(key for key in schema if key == CONF_TOKEN)
|
||||
assert token_key.default() == DEFAULT_TOKEN
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("method", "error", "expected"),
|
||||
[
|
||||
("get_settings", NinlightAuthError("401"), "invalid_auth"),
|
||||
("get_settings", NinlightConnectionError("down"), "cannot_connect"),
|
||||
("get_info", NinlightResponseError("no JSON"), "not_ninlight"),
|
||||
("get_settings", RuntimeError("boom"), "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_user_flow_errors_then_recovers(
|
||||
hass: HomeAssistant, mock_client: MagicMock, method: str, error: Exception, expected: str
|
||||
) -> None:
|
||||
mock = getattr(mock_client, method)
|
||||
original = mock.side_effect
|
||||
mock.side_effect = error
|
||||
result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_HOST: HOST_RGB, CONF_TOKEN: TOKEN}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": expected}
|
||||
mock.side_effect = original
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_HOST: HOST_RGB, CONF_TOKEN: TOKEN}
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_user_flow_same_lamp_under_new_address_updates_host(
|
||||
hass: HomeAssistant, mock_client: MagicMock
|
||||
) -> None:
|
||||
entry = existing_entry(hass)
|
||||
result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_HOST: "192.168.2.99", CONF_TOKEN: TOKEN}
|
||||
)
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
assert entry.data[CONF_HOST] == "192.168.2.99"
|
||||
|
||||
|
||||
async def test_zeroconf_flow_creates_entry(hass: HomeAssistant, mock_client: MagicMock) -> None:
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=zeroconf_info()
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "zeroconf_confirm"
|
||||
assert result["description_placeholders"] == {"name": "gu10_rgb", "host": HOST_RGB}
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"], {CONF_TOKEN: TOKEN})
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["data"] == {CONF_HOST: HOST_RGB, CONF_TOKEN: TOKEN}
|
||||
|
||||
|
||||
async def test_zeroconf_uses_announced_name(hass: HomeAssistant, mock_client: MagicMock) -> None:
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=zeroconf_info(name="Flur")
|
||||
)
|
||||
assert result["description_placeholders"]["name"] == "Flur"
|
||||
|
||||
|
||||
async def test_zeroconf_known_lamp_with_new_ip_updates_host(
|
||||
hass: HomeAssistant, mock_client: MagicMock
|
||||
) -> None:
|
||||
entry = existing_entry(hass)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=zeroconf_info(ip="192.168.2.77")
|
||||
)
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
assert entry.data[CONF_HOST] == "192.168.2.77"
|
||||
mock_client.get_settings.assert_not_awaited()
|
||||
|
||||
|
||||
async def test_zeroconf_without_mac_is_not_ninlight(
|
||||
hass: HomeAssistant, mock_client: MagicMock
|
||||
) -> None:
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=zeroconf_info(mac=None)
|
||||
)
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "not_ninlight"
|
||||
|
||||
|
||||
async def test_zeroconf_confirm_shows_auth_error(
|
||||
hass: HomeAssistant, mock_client: MagicMock
|
||||
) -> None:
|
||||
mock_client.get_settings.side_effect = NinlightAuthError("401")
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=zeroconf_info()
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_TOKEN: "falsch123"}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": "invalid_auth"}
|
||||
|
||||
|
||||
async def test_reauth_updates_token(hass: HomeAssistant, mock_client: MagicMock) -> None:
|
||||
entry = existing_entry(hass)
|
||||
result = await entry.start_reauth_flow(hass)
|
||||
assert result["step_id"] == "reauth_confirm"
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_TOKEN: NEW_TOKEN}
|
||||
)
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "reauth_successful"
|
||||
assert entry.data == {CONF_HOST: HOST_RGB, CONF_TOKEN: NEW_TOKEN}
|
||||
|
||||
|
||||
async def test_reauth_rejects_other_lamp(
|
||||
hass: HomeAssistant, mock_client: MagicMock, lamp_info: dict[str, Any]
|
||||
) -> None:
|
||||
entry = existing_entry(hass)
|
||||
lamp_info["mac"] = INFO_CW["mac"]
|
||||
result = await entry.start_reauth_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_TOKEN: NEW_TOKEN}
|
||||
)
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "wrong_device"
|
||||
assert entry.data[CONF_TOKEN] == TOKEN
|
||||
|
||||
|
||||
async def test_reauth_shows_error_for_wrong_token(
|
||||
hass: HomeAssistant, mock_client: MagicMock
|
||||
) -> None:
|
||||
entry = existing_entry(hass)
|
||||
mock_client.get_settings.side_effect = NinlightAuthError("401")
|
||||
result = await entry.start_reauth_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_TOKEN: "falsch123"}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": "invalid_auth"}
|
||||
|
||||
|
||||
async def test_reconfigure_updates_host_and_token(
|
||||
hass: HomeAssistant, mock_client: MagicMock
|
||||
) -> None:
|
||||
entry = existing_entry(hass)
|
||||
result = await entry.start_reconfigure_flow(hass)
|
||||
assert result["step_id"] == "reconfigure"
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_HOST: "192.168.2.40", CONF_TOKEN: NEW_TOKEN}
|
||||
)
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "reconfigure_successful"
|
||||
assert entry.data == {CONF_HOST: "192.168.2.40", CONF_TOKEN: NEW_TOKEN}
|
||||
|
||||
|
||||
async def test_reconfigure_rejects_other_lamp(
|
||||
hass: HomeAssistant, mock_client: MagicMock, lamp_info: dict[str, Any]
|
||||
) -> None:
|
||||
entry = existing_entry(hass)
|
||||
lamp_info["mac"] = INFO_CW["mac"]
|
||||
result = await entry.start_reconfigure_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_HOST: "192.168.2.40", CONF_TOKEN: TOKEN}
|
||||
)
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "wrong_device"
|
||||
assert entry.data[CONF_HOST] == HOST_RGB
|
||||
|
||||
|
||||
async def test_reconfigure_shows_connection_error(
|
||||
hass: HomeAssistant, mock_client: MagicMock
|
||||
) -> None:
|
||||
entry = existing_entry(hass)
|
||||
mock_client.get_settings.side_effect = NinlightConnectionError("down")
|
||||
result = await entry.start_reconfigure_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_HOST: "192.168.2.40", CONF_TOKEN: TOKEN}
|
||||
)
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
async def test_options_store_interval_and_normalized_base_url(hass: HomeAssistant) -> None:
|
||||
entry = existing_entry(hass)
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
{CONF_SCAN_INTERVAL: 45, CONF_LAMP_BASE_URL: " http://192.168.1.50:8123/ "},
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert entry.options == {CONF_SCAN_INTERVAL: 45, CONF_LAMP_BASE_URL: "http://192.168.1.50:8123"}
|
||||
|
||||
|
||||
async def test_options_empty_base_url_is_not_stored(hass: HomeAssistant) -> None:
|
||||
entry = existing_entry(hass)
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"], {CONF_SCAN_INTERVAL: 30, CONF_LAMP_BASE_URL: ""}
|
||||
)
|
||||
assert entry.options == {CONF_SCAN_INTERVAL: 30}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("url", "error"),
|
||||
[
|
||||
("https://192.168.1.50:8123", "invalid_base_url"),
|
||||
("http://" + "a" * 183, "base_url_too_long"),
|
||||
],
|
||||
)
|
||||
async def test_options_reject_unusable_base_url(hass: HomeAssistant, url: str, error: str) -> None:
|
||||
entry = existing_entry(hass)
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"], {CONF_SCAN_INTERVAL: 30, CONF_LAMP_BASE_URL: url}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {CONF_LAMP_BASE_URL: error}
|
||||
|
||||
|
||||
async def test_options_reject_interval_out_of_range(hass: HomeAssistant) -> None:
|
||||
entry = existing_entry(hass)
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
with pytest.raises(InvalidData):
|
||||
await hass.config_entries.options.async_configure(
|
||||
result["flow_id"], {CONF_SCAN_INTERVAL: 5}
|
||||
)
|
||||
@@ -0,0 +1,52 @@
|
||||
"""A rejected token during operation opens the re-authentication flow."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_REAUTH
|
||||
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_ON
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
import pytest
|
||||
from pytest_homeassistant_custom_component.common import async_fire_time_changed
|
||||
|
||||
from custom_components.ninlight.api import NinlightAuthError
|
||||
from custom_components.ninlight.const import DOMAIN
|
||||
from tests.helpers import entity_id_for
|
||||
|
||||
|
||||
def reauth_flows(hass: HomeAssistant) -> list[str]:
|
||||
return [
|
||||
flow["context"]["source"]
|
||||
for flow in hass.config_entries.flow.async_progress_by_handler(DOMAIN)
|
||||
]
|
||||
|
||||
|
||||
async def test_rejected_write_starts_reauth(
|
||||
hass: HomeAssistant, init_integration, mock_client: MagicMock
|
||||
) -> None:
|
||||
mock_client.set_state.side_effect = NinlightAuthError("401")
|
||||
with pytest.raises(HomeAssistantError) as raised:
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: entity_id_for(hass, "light", "light")},
|
||||
blocking=True,
|
||||
)
|
||||
assert raised.value.translation_key == "auth_failed"
|
||||
await hass.async_block_till_done()
|
||||
assert reauth_flows(hass) == [SOURCE_REAUTH]
|
||||
|
||||
|
||||
async def test_rejected_poll_starts_reauth(
|
||||
hass: HomeAssistant, init_integration, mock_client: MagicMock, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
mock_client.get_state.side_effect = NinlightAuthError("401") # lamp with auth_all
|
||||
freezer.tick(timedelta(seconds=31))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert reauth_flows(hass) == [SOURCE_REAUTH]
|
||||
Reference in New Issue
Block a user