Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add ConfigFlow for Lupusec (#108740) 

* init support for config flow for lupusec

* correctly iterate over BinarySensorDeviceClass values for device class

* bump lupupy to 0.3.2

* Updated device info for lupusec

* revert bump lupupy for separate pr

* fixed lupusec test-cases

* Change setup to async_setup

* remove redundant check for hass.data.setdefault

* init support for config flow for lupusec

* correctly iterate over BinarySensorDeviceClass values for device class

* bump lupupy to 0.3.2

* Updated device info for lupusec

* revert bump lupupy for separate pr

* fixed lupusec test-cases

* Change setup to async_setup

* remove redundant check for hass.data.setdefault

* resolve merge error lupupy

* connection check when setting up config entry

* removed unique_id and device_info for separate pr

* changed name to friendly name

* renamed LUPUSEC_PLATFORMS to PLATFORMS

* preparation for code review

* necessary changes for pr

* changed config access

* duplicate entry check

* types added for setup_entry and test_host_connection

* removed name for lupusec system

* removed config entry from LupusecDevice

* fixes for sensors

* added else block for try

* added integration warning

* pass config to config_flow

* fix test cases for new config flow

* added error strings

* changed async_create_entry invocation

* added tests for exception handling

* use parametrize

* use parametrize for tests

* recover test

* test unique id

* import from yaml tests

* import error test cases

* Update tests/components/lupusec/test_config_flow.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* fixed test case

* removed superfluous test cases

* self._async_abort_entries_match added

* lib patching call

* _async_abort_entries_match

* patch lupupy lib instead of test connection

* removed statements

* test_flow_source_import_already_configured

* Update homeassistant/components/lupusec/config_flow.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* removed unique_id from mockentry

* added __init__.py to .coveragerc

---------

Co-authored-by: suaveolent <suaveolent@users.noreply.github.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
pull/108863/head^2
suaveolent 2024-01-25 18:52:30 +01:00 committed by GitHub
parent bb8828c86f
commit faad9a7584
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 510 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.coveragerc
View File

@ -705,7 +705,10 @@ omit =
homeassistant/components/loqed/sensor.py
homeassistant/components/luci/device_tracker.py
homeassistant/components/luftdaten/sensor.py
homeassistant/components/lupusec/*
homeassistant/components/lupusec/__init__.py
homeassistant/components/lupusec/alarm_control_panel.py
homeassistant/components/lupusec/binary_sensor.py
homeassistant/components/lupusec/switch.py
homeassistant/components/lutron/__init__.py
homeassistant/components/lutron/binary_sensor.py
homeassistant/components/lutron/cover.py

3
CODEOWNERS
View File

@ -757,7 +757,8 @@ build.json @home-assistant/supervisor
/homeassistant/components/luci/ @mzdrale
/homeassistant/components/luftdaten/ @fabaff @frenck
/tests/components/luftdaten/ @fabaff @frenck
/homeassistant/components/lupusec/ @majuss
/homeassistant/components/lupusec/ @majuss @suaveolent
/tests/components/lupusec/ @majuss @suaveolent
/homeassistant/components/lutron/ @cdheiser @wilburCForce
/tests/components/lutron/ @cdheiser @wilburCForce
/homeassistant/components/lutron_caseta/ @swails @bdraco @danaues

107
homeassistant/components/lupusec/__init__.py
View File

@ -5,19 +5,24 @@ import lupupy
from lupupy.exceptions import LupusecException
import voluptuous as vol
from homeassistant.components import persistent_notification
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_IP_ADDRESS,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, discovery
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType
from .const import INTEGRATION_TITLE, ISSUE_PLACEHOLDER
_LOGGER = logging.getLogger(__name__)
DOMAIN = "lupusec"
@ -39,36 +44,91 @@ CONFIG_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA,
)
LUPUSEC_PLATFORMS = [
PLATFORMS: list[Platform] = [
Platform.ALARM_CONTROL_PANEL,
Platform.BINARY_SENSOR,
Platform.SWITCH,
]
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Lupusec component."""
async def handle_async_init_result(hass: HomeAssistant, domain: str, conf: dict):
"""Handle the result of the async_init to issue deprecated warnings."""
flow = hass.config_entries.flow
result = await flow.async_init(domain, context={"source": SOURCE_IMPORT}, data=conf)
if (
result["type"] == FlowResultType.CREATE_ENTRY
or result["reason"] == "already_configured"
):
async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2024.8.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": INTEGRATION_TITLE,
},
)
else:
async_create_issue(
hass,
DOMAIN,
f"deprecated_yaml_import_issue_${result['reason']}",
breaks_in_ha_version="2024.8.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key=f"deprecated_yaml_import_issue_${result['reason']}",
translation_placeholders=ISSUE_PLACEHOLDER,
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the lupusec integration."""
if DOMAIN not in config:
return True
conf = config[DOMAIN]
username = conf[CONF_USERNAME]
password = conf[CONF_PASSWORD]
ip_address = conf[CONF_IP_ADDRESS]
name = conf.get(CONF_NAME)
hass.async_create_task(handle_async_init_result(hass, DOMAIN, conf))
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up this integration using UI."""
host = entry.data[CONF_HOST]
username = entry.data[CONF_USERNAME]
password = entry.data[CONF_PASSWORD]
try:
hass.data[DOMAIN] = LupusecSystem(username, password, ip_address, name)
except LupusecException as ex:
_LOGGER.error(ex)
persistent_notification.create(
hass,
f"Error: {ex}<br />You will need to restart hass after fixing.",
title=NOTIFICATION_TITLE,
notification_id=NOTIFICATION_ID,
lupusec_system = await hass.async_add_executor_job(
LupusecSystem,
username,
password,
host,
)
except LupusecException:
_LOGGER.error("Failed to connect to Lupusec device at %s", host)
return False
except Exception as ex: # pylint: disable=broad-except
_LOGGER.error(
"Unknown error while trying to connect to Lupusec device at %s: %s",
host,
ex,
)
return False
for platform in LUPUSEC_PLATFORMS:
discovery.load_platform(hass, platform, DOMAIN, {}, config)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = lupusec_system
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
@ -76,16 +136,15 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool:
class LupusecSystem:
"""Lupusec System class."""
def __init__(self, username, password, ip_address, name):
def __init__(self, username, password, ip_address) -> None:
"""Initialize the system."""
self.lupusec = lupupy.Lupusec(username, password, ip_address)
self.name = name
class LupusecDevice(Entity):
"""Representation of a Lupusec device."""
def __init__(self, data, device):
def __init__(self, data, device) -> None:
"""Initialize a sensor for Lupusec device."""
self._data = data
self._device = device

16
homeassistant/components/lupusec/alarm_control_panel.py
View File

@ -7,6 +7,7 @@ from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntity,
AlarmControlPanelEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_HOME,
@ -15,28 +16,23 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DOMAIN as LUPUSEC_DOMAIN, LupusecDevice
SCAN_INTERVAL = timedelta(seconds=2)
def setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
config_entry: ConfigEntry,
async_add_devices: AddEntitiesCallback,
) -> None:
"""Set up an alarm control panel for a Lupusec device."""
if discovery_info is None:
return
data = hass.data[LUPUSEC_DOMAIN]
data = hass.data[LUPUSEC_DOMAIN][config_entry.entry_id]
alarm_devices = [LupusecAlarm(data, data.lupusec.get_alarm())]
add_entities(alarm_devices)
async_add_devices(alarm_devices)
class LupusecAlarm(LupusecDevice, AlarmControlPanelEntity):

35
homeassistant/components/lupusec/binary_sensor.py
View File

@ -2,38 +2,41 @@
from __future__ import annotations
from datetime import timedelta
import logging
import lupupy.constants as CONST
from homeassistant.components.binary_sensor import DEVICE_CLASSES, BinarySensorEntity
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DOMAIN as LUPUSEC_DOMAIN, LupusecDevice
from . import DOMAIN, LupusecDevice
SCAN_INTERVAL = timedelta(seconds=2)
_LOGGER = logging.getLogger(__name__)
def setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
config_entry: ConfigEntry,
async_add_devices: AddEntitiesCallback,
) -> None:
"""Set up a sensor for an Lupusec device."""
if discovery_info is None:
return
"""Set up a binary sensors for a Lupusec device."""
data = hass.data[LUPUSEC_DOMAIN]
data = hass.data[DOMAIN][config_entry.entry_id]
device_types = CONST.TYPE_OPENING + CONST.TYPE_SENSOR
devices = []
sensors = []
for device in data.lupusec.get_devices(generic_type=device_types):
devices.append(LupusecBinarySensor(data, device))
sensors.append(LupusecBinarySensor(data, device))
add_entities(devices)
async_add_devices(sensors)
class LupusecBinarySensor(LupusecDevice, BinarySensorEntity):
@ -47,6 +50,8 @@ class LupusecBinarySensor(LupusecDevice, BinarySensorEntity):
@property
def device_class(self):
"""Return the class of the binary sensor."""
if self._device.generic_type not in DEVICE_CLASSES:
if self._device.generic_type not in (
item.value for item in BinarySensorDeviceClass
):
return None
return self._device.generic_type

110
homeassistant/components/lupusec/config_flow.py Normal file
View File

@ -0,0 +1,110 @@
""""Config flow for Lupusec integration."""
import logging
from typing import Any
import lupupy
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import (
CONF_HOST,
CONF_IP_ADDRESS,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
)
class LupusecConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Lupusec config flow."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
errors = {}
if user_input is not None:
self._async_abort_entries_match(user_input)
host = user_input[CONF_HOST]
username = user_input[CONF_USERNAME]
password = user_input[CONF_PASSWORD]
try:
await test_host_connection(self.hass, host, username, password)
except CannotConnect:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return self.async_create_entry(
title=host,
data=user_input,
)
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA, errors=errors
)
async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
"""Import the yaml config."""
self._async_abort_entries_match(
{
CONF_HOST: user_input[CONF_IP_ADDRESS],
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
}
)
host = user_input[CONF_IP_ADDRESS]
username = user_input[CONF_USERNAME]
password = user_input[CONF_PASSWORD]
try:
await test_host_connection(self.hass, host, username, password)
except CannotConnect:
return self.async_abort(reason="cannot_connect")
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
return self.async_create_entry(
title=user_input.get(CONF_NAME, host),
data={
CONF_HOST: host,
CONF_USERNAME: username,
CONF_PASSWORD: password,
},
)
async def test_host_connection(
hass: HomeAssistant, host: str, username: str, password: str
):
"""Test if the host is reachable and is actually a Lupusec device."""
try:
await hass.async_add_executor_job(lupupy.Lupusec, username, password, host)
except lupupy.LupusecException:
_LOGGER.error("Failed to connect to Lupusec device at %s", host)
raise CannotConnect
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""

6
homeassistant/components/lupusec/const.py Normal file
View File

@ -0,0 +1,6 @@
"""Constants for the Lupusec component."""
DOMAIN = "lupusec"
INTEGRATION_TITLE = "Lupus Electronics LUPUSEC"
ISSUE_PLACEHOLDER = {"url": "/config/integrations/dashboard/add?domain=lupusec"}

3
homeassistant/components/lupusec/manifest.json
View File

@ -1,7 +1,8 @@
{
"domain": "lupusec",
"name": "Lupus Electronics LUPUSEC",
"codeowners": ["@majuss"],
"codeowners": ["@majuss", "@suaveolent"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/lupusec",
"iot_class": "local_polling",
"loggers": ["lupupy"],

31
homeassistant/components/lupusec/strings.json Normal file
View File

@ -0,0 +1,31 @@
{
"config": {
"step": {
"user": {
"title": "Lupus Electronics LUPUSEC connection",
"data": {
"host": "[%key:common::config_flow::data::host%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"issues": {
"deprecated_yaml_import_issue_cannot_connect": {
"title": "The Lupus Electronics LUPUSEC YAML configuration import failed",
"description": "Configuring Lupus Electronics LUPUSEC using YAML is being removed but there was an connection error importing your YAML configuration.\n\nEnsure connection to Lupus Electronics LUPUSEC works and restart Home Assistant to try again or remove the Lupus Electronics LUPUSEC YAML configuration from your configuration.yaml file and continue to [set up the integration]({url}) manually."
},
"deprecated_yaml_import_issue_unknown": {
"title": "The Lupus Electronics LUPUSEC YAML configuration import failed",
"description": "Configuring Lupus Electronics LUPUSEC using YAML is being removed but there was an unknown error when trying to import the YAML configuration.\n\nEnsure the imported configuration is correct and remove the Lupus Electronics LUPUSEC YAML configuration from your configuration.yaml file and continue to [set up the integration]({url}) manually."
}
}
}

21
homeassistant/components/lupusec/switch.py
View File

@ -7,34 +7,31 @@ from typing import Any
import lupupy.constants as CONST
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DOMAIN as LUPUSEC_DOMAIN, LupusecDevice
from . import DOMAIN, LupusecDevice
SCAN_INTERVAL = timedelta(seconds=2)
def setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
config_entry: ConfigEntry,
async_add_devices: AddEntitiesCallback,
) -> None:
"""Set up Lupusec switch devices."""
if discovery_info is None:
return
data = hass.data[LUPUSEC_DOMAIN]
data = hass.data[DOMAIN][config_entry.entry_id]
device_types = CONST.TYPE_SWITCH
devices = []
switches = []
for device in data.lupusec.get_devices(generic_type=device_types):
devices.append(LupusecSwitch(data, device))
switches.append(LupusecSwitch(data, device))
add_entities(devices)
async_add_devices(switches)
class LupusecSwitch(LupusecDevice, SwitchEntity):

1
homeassistant/generated/config_flows.py
View File

@ -289,6 +289,7 @@ FLOWS = {
"lookin",
"loqed",
"luftdaten",
"lupusec",
"lutron",
"lutron_caseta",
"lyric",

2
homeassistant/generated/integrations.json
View File

@ -3312,7 +3312,7 @@
"lupusec": {
"name": "Lupus Electronics LUPUSEC",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "local_polling"
},
"lutron": {

3
requirements_test_all.txt
View File

@ -979,6 +979,9 @@ loqedAPI==2.1.8
# homeassistant.components.luftdaten
luftdaten==0.7.4
# homeassistant.components.lupusec
lupupy==0.3.2
# homeassistant.components.scrape
lxml==5.1.0

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

@ -0,0 +1 @@
"""Define tests for the lupusec component."""

231
tests/components/lupusec/test_config_flow.py Normal file
View File

@ -0,0 +1,231 @@
""""Unit tests for the Lupusec config flow."""
from unittest.mock import patch
from lupupy import LupusecException
import pytest
from homeassistant import config_entries
from homeassistant.components.lupusec.const import DOMAIN
from homeassistant.const import (
CONF_HOST,
CONF_IP_ADDRESS,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
MOCK_DATA_STEP = {
CONF_HOST: "test-host.lan",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
}
MOCK_IMPORT_STEP = {
CONF_IP_ADDRESS: "test-host.lan",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
}
MOCK_IMPORT_STEP_NAME = {
CONF_IP_ADDRESS: "test-host.lan",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_NAME: "test-name",
}
async def test_form_valid_input(hass: HomeAssistant) -> None:
"""Test handling valid user input."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.lupusec.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.lupusec.config_flow.lupupy.Lupusec.__init__",
return_value=None,
) as mock_initialize_lupusec:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_DATA_STEP,
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == MOCK_DATA_STEP[CONF_HOST]
assert result2["data"] == MOCK_DATA_STEP
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_initialize_lupusec.mock_calls) == 1
@pytest.mark.parametrize(
("raise_error", "text_error"),
[
(LupusecException("Test lupusec exception"), "cannot_connect"),
(Exception("Test unknown exception"), "unknown"),
],
)
async def test_flow_user_init_data_error_and_recover(
hass: HomeAssistant, raise_error, text_error
) -> None:
"""Test exceptions and recovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.lupusec.config_flow.lupupy.Lupusec.__init__",
side_effect=raise_error,
) as mock_initialize_lupusec:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_DATA_STEP,
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": text_error}
assert len(mock_initialize_lupusec.mock_calls) == 1
# Recover
with patch(
"homeassistant.components.lupusec.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.lupusec.config_flow.lupupy.Lupusec.__init__",
return_value=None,
) as mock_initialize_lupusec:
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_DATA_STEP,
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == MOCK_DATA_STEP[CONF_HOST]
assert result3["data"] == MOCK_DATA_STEP
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_initialize_lupusec.mock_calls) == 1
async def test_flow_user_init_data_already_configured(hass: HomeAssistant) -> None:
"""Test duplicate config entry.."""
entry = MockConfigEntry(
domain=DOMAIN,
title=MOCK_DATA_STEP[CONF_HOST],
data=MOCK_DATA_STEP,
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {}
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_DATA_STEP,
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "already_configured"
@pytest.mark.parametrize(
("mock_import_step", "mock_title"),
[
(MOCK_IMPORT_STEP, MOCK_IMPORT_STEP[CONF_IP_ADDRESS]),
(MOCK_IMPORT_STEP_NAME, MOCK_IMPORT_STEP_NAME[CONF_NAME]),
],
)
async def test_flow_source_import(
hass: HomeAssistant, mock_import_step, mock_title
) -> None:
"""Test configuration import from YAML."""
with patch(
"homeassistant.components.lupusec.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.lupusec.config_flow.lupupy.Lupusec.__init__",
return_value=None,
) as mock_initialize_lupusec:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=mock_import_step,
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == mock_title
assert result["data"] == MOCK_DATA_STEP
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_initialize_lupusec.mock_calls) == 1
@pytest.mark.parametrize(
("raise_error", "text_error"),
[
(LupusecException("Test lupusec exception"), "cannot_connect"),
(Exception("Test unknown exception"), "unknown"),
],
)
async def test_flow_source_import_error_and_recover(
hass: HomeAssistant, raise_error, text_error
) -> None:
"""Test exceptions and recovery."""
with patch(
"homeassistant.components.lupusec.config_flow.lupupy.Lupusec.__init__",
side_effect=raise_error,
) as mock_initialize_lupusec:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=MOCK_IMPORT_STEP,
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == text_error
assert len(mock_initialize_lupusec.mock_calls) == 1
async def test_flow_source_import_already_configured(hass: HomeAssistant) -> None:
"""Test duplicate config entry.."""
entry = MockConfigEntry(
domain=DOMAIN,
title=MOCK_DATA_STEP[CONF_HOST],
data=MOCK_DATA_STEP,
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=MOCK_IMPORT_STEP,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"