Add D-Link config flow (#84927)

This commit is contained in:
Robert Hillis 2023-01-10 19:10:56 -05:00 committed by GitHub
parent 1afb30344a
commit 3d7c61bbed
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 535 additions and 90 deletions

View File

@ -230,6 +230,9 @@ omit =
homeassistant/components/discord/notify.py homeassistant/components/discord/notify.py
homeassistant/components/dlib_face_detect/image_processing.py homeassistant/components/dlib_face_detect/image_processing.py
homeassistant/components/dlib_face_identify/image_processing.py homeassistant/components/dlib_face_identify/image_processing.py
homeassistant/components/dlink/__init__.py
homeassistant/components/dlink/data.py
homeassistant/components/dlink/entity.py
homeassistant/components/dlink/switch.py homeassistant/components/dlink/switch.py
homeassistant/components/dominos/* homeassistant/components/dominos/*
homeassistant/components/doods/* homeassistant/components/doods/*

View File

@ -265,6 +265,8 @@ build.json @home-assistant/supervisor
/tests/components/discord/ @tkdrob /tests/components/discord/ @tkdrob
/homeassistant/components/discovery/ @home-assistant/core /homeassistant/components/discovery/ @home-assistant/core
/tests/components/discovery/ @home-assistant/core /tests/components/discovery/ @home-assistant/core
/homeassistant/components/dlink/ @tkdrob
/tests/components/dlink/ @tkdrob
/homeassistant/components/dlna_dmr/ @StevenLooman @chishm /homeassistant/components/dlna_dmr/ @StevenLooman @chishm
/tests/components/dlna_dmr/ @StevenLooman @chishm /tests/components/dlna_dmr/ @StevenLooman @chishm
/homeassistant/components/dlna_dms/ @chishm /homeassistant/components/dlna_dms/ @chishm

View File

@ -1 +1,39 @@
"""The dlink component.""" """The D-Link Power Plug integration."""
from __future__ import annotations
from pyW215.pyW215 import SmartPlug
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .const import CONF_USE_LEGACY_PROTOCOL, DOMAIN
from .data import SmartPlugData
PLATFORMS = [Platform.SWITCH]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up D-Link Power Plug from a config entry."""
smartplug = await hass.async_add_executor_job(
SmartPlug,
entry.data[CONF_HOST],
entry.data[CONF_PASSWORD],
entry.data[CONF_USERNAME],
entry.data[CONF_USE_LEGACY_PROTOCOL],
)
if not smartplug.authenticated and entry.data[CONF_USE_LEGACY_PROTOCOL]:
raise ConfigEntryNotReady("Cannot connect/authenticate")
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = SmartPlugData(smartplug)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

View File

@ -0,0 +1,80 @@
"""Config flow for the D-Link Power Plug integration."""
from __future__ import annotations
import logging
from typing import Any
from pyW215.pyW215 import SmartPlug
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.data_entry_flow import FlowResult
from .const import CONF_USE_LEGACY_PROTOCOL, DEFAULT_NAME, DEFAULT_USERNAME, DOMAIN
_LOGGER = logging.getLogger(__name__)
class DLinkFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for D-Link Power Plug."""
async def async_step_import(self, config: dict[str, Any]) -> FlowResult:
"""Import a config entry."""
self._async_abort_entries_match({CONF_HOST: config[CONF_HOST]})
title = config.pop(CONF_NAME, DEFAULT_NAME)
return self.async_create_entry(
title=title,
data=config,
)
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({CONF_HOST: user_input[CONF_HOST]})
error = await self.hass.async_add_executor_job(
self._try_connect, user_input
)
if error is None:
return self.async_create_entry(
title=DEFAULT_NAME,
data=user_input,
)
errors["base"] = error
user_input = user_input or {}
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST, default=user_input.get(CONF_HOST, "")): str,
vol.Optional(
CONF_USERNAME,
default=user_input.get(CONF_USERNAME, DEFAULT_USERNAME),
): str,
vol.Required(CONF_PASSWORD): str,
vol.Required(CONF_USE_LEGACY_PROTOCOL): bool,
}
),
errors=errors,
)
def _try_connect(self, user_input: dict[str, Any]) -> str | None:
"""Try connecting to D-Link Power Plug."""
try:
smartplug = SmartPlug(
user_input[CONF_HOST],
user_input[CONF_PASSWORD],
user_input[CONF_USERNAME],
user_input[CONF_USE_LEGACY_PROTOCOL],
)
except Exception as ex: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception: %s", ex)
return "unknown"
if smartplug.authenticated:
return None
return "cannot_connect"

View File

@ -0,0 +1,12 @@
"""Constants for the D-Link Power Plug integration."""
ATTRIBUTION = "Data provided by D-Link"
ATTR_TOTAL_CONSUMPTION = "total_consumption"
CONF_USE_LEGACY_PROTOCOL = "use_legacy_protocol"
DEFAULT_NAME = "D-Link Smart Plug W215"
DEFAULT_USERNAME = "admin"
DOMAIN = "dlink"
MANUFACTURER = "D-Link"

View File

@ -0,0 +1,57 @@
"""Data for the D-Link Power Plug integration."""
from __future__ import annotations
from datetime import datetime
import logging
import urllib
from pyW215.pyW215 import SmartPlug
from homeassistant.util import dt as dt_util
_LOGGER = logging.getLogger(__name__)
class SmartPlugData:
"""Get the latest data from smart plug."""
def __init__(self, smartplug: SmartPlug) -> None:
"""Initialize the data object."""
self.smartplug = smartplug
self.state: str | None = None
self.temperature: float | None = None
self.current_consumption = None
self.total_consumption: str | None = None
self.available = False
self._n_tried = 0
self._last_tried: datetime | None = None
def update(self) -> None:
"""Get the latest data from the smart plug."""
if self._last_tried is not None:
last_try_s = (dt_util.now() - self._last_tried).total_seconds() / 60
retry_seconds = min(self._n_tried * 2, 10) - last_try_s
if self._n_tried > 0 and retry_seconds > 0:
_LOGGER.warning("Waiting %s s to retry", retry_seconds)
return
_state = "unknown"
try:
self._last_tried = dt_util.now()
_state = self.smartplug.state
except urllib.error.HTTPError:
_LOGGER.error("D-Link connection problem")
if _state == "unknown":
self._n_tried += 1
self.available = False
_LOGGER.warning("Failed to connect to D-Link switch")
return
self.state = _state
self.available = True
self.temperature = self.smartplug.temperature
self.current_consumption = self.smartplug.current_consumption
self.total_consumption = self.smartplug.total_consumption
self._n_tried = 0

View File

@ -0,0 +1,41 @@
"""Entity representing a D-Link Power Plug device."""
from __future__ import annotations
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import ATTR_CONNECTIONS
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity import DeviceInfo, Entity, EntityDescription
from .const import ATTRIBUTION, DOMAIN, MANUFACTURER
from .data import SmartPlugData
class DLinkEntity(Entity):
"""Representation of a D-Link Power Plug entity."""
_attr_attribution = ATTRIBUTION
def __init__(
self,
data: SmartPlugData,
config_entry: ConfigEntry,
description: EntityDescription,
) -> None:
"""Initialize a D-Link Power Plug entity."""
self.data = data
self.entity_description = description
if config_entry.source == SOURCE_IMPORT:
self._attr_name = config_entry.title
else:
self._attr_name = f"{config_entry.title} {description.key}"
self._attr_unique_id = f"{config_entry.entry_id}_{description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, config_entry.entry_id)},
manufacturer=MANUFACTURER,
model=data.smartplug.model_name,
name=config_entry.title,
)
if config_entry.unique_id:
self._attr_device_info[ATTR_CONNECTIONS] = {
(dr.CONNECTION_NETWORK_MAC, config_entry.unique_id)
}

View File

@ -1,9 +1,11 @@
{ {
"domain": "dlink", "domain": "dlink",
"name": "D-Link Wi-Fi Smart Plugs", "name": "D-Link Wi-Fi Smart Plugs",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/dlink", "documentation": "https://www.home-assistant.io/integrations/dlink",
"requirements": ["pyW215==0.7.0"], "requirements": ["pyW215==0.7.0"],
"codeowners": [], "codeowners": ["@tkdrob"],
"iot_class": "local_polling", "iot_class": "local_polling",
"loggers": ["pyW215"] "loggers": ["pyW215"],
"integration_type": "device"
} }

View File

@ -0,0 +1,27 @@
{
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"password": "Password (default: PIN code on the back)",
"username": "[%key:common::config_flow::data::username%]",
"use_legacy_protocol": "Use legacy protocol"
}
}
},
"error": {
"cannot_connect": "Failed to connect/authenticate",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"issues": {
"deprecated_yaml": {
"title": "The D-Link Smart Plug YAML configuration is being removed",
"description": "Configuring D-Link Smart Plug using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the D-Link Power Plug YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
}
}
}

View File

@ -1,15 +1,17 @@
"""Support for D-Link W215 smart switch.""" """Support for D-Link Power Plug Switches."""
from __future__ import annotations from __future__ import annotations
from datetime import timedelta from datetime import timedelta
import logging
from typing import Any from typing import Any
import urllib
from pyW215.pyW215 import SmartPlug
import voluptuous as vol import voluptuous as vol
from homeassistant.components.switch import PLATFORM_SCHEMA, SwitchEntity from homeassistant.components.switch import (
PLATFORM_SCHEMA,
SwitchEntity,
SwitchEntityDescription,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import ( from homeassistant.const import (
ATTR_TEMPERATURE, ATTR_TEMPERATURE,
CONF_HOST, CONF_HOST,
@ -21,31 +23,36 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util
_LOGGER = logging.getLogger(__name__) from .const import (
ATTR_TOTAL_CONSUMPTION,
ATTR_TOTAL_CONSUMPTION = "total_consumption" CONF_USE_LEGACY_PROTOCOL,
DEFAULT_NAME,
CONF_USE_LEGACY_PROTOCOL = "use_legacy_protocol" DEFAULT_USERNAME,
DOMAIN,
DEFAULT_NAME = "D-Link Smart Plug W215" )
DEFAULT_PASSWORD = "" from .data import SmartPlugData
DEFAULT_USERNAME = "admin" from .entity import DLinkEntity
SCAN_INTERVAL = timedelta(minutes=2) SCAN_INTERVAL = timedelta(minutes=2)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{ {
vol.Required(CONF_HOST): cv.string, vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_PASSWORD, default=DEFAULT_PASSWORD): cv.string, vol.Required(CONF_PASSWORD, default=""): cv.string,
vol.Required(CONF_USERNAME, default=DEFAULT_USERNAME): cv.string, vol.Required(CONF_USERNAME, default=DEFAULT_USERNAME): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_USE_LEGACY_PROTOCOL, default=False): cv.boolean, vol.Optional(CONF_USE_LEGACY_PROTOCOL, default=False): cv.boolean,
} }
) )
SWITCH_TYPE = SwitchEntityDescription(
key="switch",
name="Switch",
)
def setup_platform( def setup_platform(
hass: HomeAssistant, hass: HomeAssistant,
@ -54,46 +61,68 @@ def setup_platform(
discovery_info: DiscoveryInfoType | None = None, discovery_info: DiscoveryInfoType | None = None,
) -> None: ) -> None:
"""Set up a D-Link Smart Plug.""" """Set up a D-Link Smart Plug."""
async_create_issue(
host = config[CONF_HOST] hass,
username = config[CONF_USERNAME] DOMAIN,
password = config[CONF_PASSWORD] "deprecated_yaml",
use_legacy_protocol = config[CONF_USE_LEGACY_PROTOCOL] breaks_in_ha_version="2023.3.0",
name = config[CONF_NAME] is_fixable=False,
severity=IssueSeverity.WARNING,
smartplug = SmartPlug(host, password, username, use_legacy_protocol) translation_key="deprecated_yaml",
data = SmartPlugData(smartplug) )
hass.async_create_task(
add_entities([SmartPlugSwitch(hass, data, name)], True) hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
)
class SmartPlugSwitch(SwitchEntity): async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the D-Link Power Plug switch."""
async_add_entities(
[
SmartPlugSwitch(
hass,
entry,
hass.data[DOMAIN][entry.entry_id],
SWITCH_TYPE,
),
],
True,
)
class SmartPlugSwitch(DLinkEntity, SwitchEntity):
"""Representation of a D-Link Smart Plug switch.""" """Representation of a D-Link Smart Plug switch."""
def __init__(self, hass, data, name): def __init__(
self,
hass: HomeAssistant,
entry: ConfigEntry,
data: SmartPlugData,
description: SwitchEntityDescription,
) -> None:
"""Initialize the switch.""" """Initialize the switch."""
super().__init__(data, entry, description)
self.units = hass.config.units self.units = hass.config.units
self.data = data
self._name = name
@property @property
def name(self): def extra_state_attributes(self) -> dict[str, Any]:
"""Return the name of the Smart Plug."""
return self._name
@property
def extra_state_attributes(self):
"""Return the state attributes of the device.""" """Return the state attributes of the device."""
try: try:
ui_temp = self.units.temperature( ui_temp = self.units.temperature(
int(self.data.temperature), UnitOfTemperature.CELSIUS int(self.data.temperature or 0), UnitOfTemperature.CELSIUS
) )
temperature = ui_temp temperature = ui_temp
except (ValueError, TypeError): except (ValueError, TypeError):
temperature = None temperature = None
try: try:
total_consumption = float(self.data.total_consumption) total_consumption = float(self.data.total_consumption or "0")
except (ValueError, TypeError): except (ValueError, TypeError):
total_consumption = None total_consumption = None
@ -105,7 +134,7 @@ class SmartPlugSwitch(SwitchEntity):
return attrs return attrs
@property @property
def is_on(self): def is_on(self) -> bool:
"""Return true if switch is on.""" """Return true if switch is on."""
return self.data.state == "ON" return self.data.state == "ON"
@ -125,48 +154,3 @@ class SmartPlugSwitch(SwitchEntity):
def available(self) -> bool: def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return self.data.available return self.data.available
class SmartPlugData:
"""Get the latest data from smart plug."""
def __init__(self, smartplug):
"""Initialize the data object."""
self.smartplug = smartplug
self.state = None
self.temperature = None
self.current_consumption = None
self.total_consumption = None
self.available = False
self._n_tried = 0
self._last_tried = None
def update(self):
"""Get the latest data from the smart plug."""
if self._last_tried is not None:
last_try_s = (dt_util.now() - self._last_tried).total_seconds() / 60
retry_seconds = min(self._n_tried * 2, 10) - last_try_s
if self._n_tried > 0 and retry_seconds > 0:
_LOGGER.warning("Waiting %s s to retry", retry_seconds)
return
_state = "unknown"
try:
self._last_tried = dt_util.now()
_state = self.smartplug.state
except urllib.error.HTTPError:
_LOGGER.error("D-Link connection problem")
if _state == "unknown":
self._n_tried += 1
self.available = False
_LOGGER.warning("Failed to connect to D-Link switch")
return
self.state = _state
self.available = True
self.temperature = self.smartplug.temperature
self.current_consumption = self.smartplug.current_consumption
self.total_consumption = self.smartplug.total_consumption
self._n_tried = 0

View File

@ -0,0 +1,27 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect/authenticate",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"host": "Host",
"password": "Password (default: PIN code on the back)",
"username": "Username",
"use_legacy_protocol": "Use legacy protocol"
}
}
}
},
"issues": {
"deprecated_yaml": {
"title": "The D-Link Smart Plug YAML configuration is being removed",
"description": "Configuring D-Link Smart Plug using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the D-Link Power Plug YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
}
}
}

View File

@ -93,6 +93,7 @@ FLOWS = {
"dialogflow", "dialogflow",
"directv", "directv",
"discord", "discord",
"dlink",
"dlna_dmr", "dlna_dmr",
"dlna_dms", "dlna_dms",
"dnsip", "dnsip",

View File

@ -1088,8 +1088,8 @@
}, },
"dlink": { "dlink": {
"name": "D-Link Wi-Fi Smart Plugs", "name": "D-Link Wi-Fi Smart Plugs",
"integration_type": "hub", "integration_type": "device",
"config_flow": false, "config_flow": true,
"iot_class": "local_polling" "iot_class": "local_polling"
}, },
"dlna": { "dlna": {

View File

@ -1063,6 +1063,9 @@ pyRFXtrx==0.30.0
# homeassistant.components.tibber # homeassistant.components.tibber
pyTibber==0.26.7 pyTibber==0.26.7
# homeassistant.components.dlink
pyW215==0.7.0
# homeassistant.components.nextbus # homeassistant.components.nextbus
py_nextbusnext==0.1.5 py_nextbusnext==0.1.5

View File

@ -0,0 +1 @@
"""Tests for the D-Link Smart Plug integration."""

View File

@ -0,0 +1,66 @@
"""Configure pytest for D-Link tests."""
from copy import deepcopy
from unittest.mock import MagicMock, patch
import pytest
from homeassistant.components.dlink.const import CONF_USE_LEGACY_PROTOCOL, DOMAIN
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
HOST = "1.2.3.4"
PASSWORD = "123456"
USERNAME = "admin"
CONF_DATA = {
CONF_HOST: HOST,
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_USE_LEGACY_PROTOCOL: True,
}
CONF_IMPORT_DATA = CONF_DATA | {CONF_NAME: "Smart Plug"}
def create_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Create fixture for adding config entry in Home Assistant."""
entry = MockConfigEntry(domain=DOMAIN, data=CONF_DATA)
entry.add_to_hass(hass)
return entry
@pytest.fixture()
def config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Add config entry in Home Assistant."""
return create_entry(hass)
@pytest.fixture()
def mocked_plug() -> MagicMock:
"""Create mocked plug device."""
mocked_plug = MagicMock()
mocked_plug.state = "OFF"
mocked_plug.temperature = 0
mocked_plug.current_consumption = "N/A"
mocked_plug.total_consumption = "N/A"
mocked_plug.authenticated = ("0123456789ABCDEF0123456789ABCDEF", "ABCDefGHiJ")
return mocked_plug
@pytest.fixture()
def mocked_plug_no_auth(mocked_plug: MagicMock) -> MagicMock:
"""Create mocked unauthenticated plug device."""
mocked_plug = deepcopy(mocked_plug)
mocked_plug.authenticated = None
return mocked_plug
def patch_config_flow(mocked_plug: MagicMock):
"""Patch D-Link Smart Plug config flow."""
return patch(
"homeassistant.components.dlink.config_flow.SmartPlug",
return_value=mocked_plug,
)

View File

@ -0,0 +1,101 @@
"""Test D-Link Smart Plug config flow."""
from unittest.mock import MagicMock, patch
from homeassistant import data_entry_flow
from homeassistant.components.dlink.const import DEFAULT_NAME, DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.core import HomeAssistant
from .conftest import CONF_DATA, CONF_IMPORT_DATA, patch_config_flow
from tests.common import MockConfigEntry
def _patch_setup_entry():
return patch("homeassistant.components.dlink.async_setup_entry")
async def test_flow_user(hass: HomeAssistant, mocked_plug: MagicMock) -> None:
"""Test user initialized flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
with patch_config_flow(mocked_plug), _patch_setup_entry():
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_DATA,
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == DEFAULT_NAME
assert result["data"] == CONF_DATA
async def test_flow_user_already_configured(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
"""Test user initialized flow with duplicate server."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_flow_user_cannot_connect(
hass: HomeAssistant, mocked_plug: MagicMock, mocked_plug_no_auth: MagicMock
) -> None:
"""Test user initialized flow with unreachable server."""
with patch_config_flow(mocked_plug_no_auth):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "cannot_connect"
with patch_config_flow(mocked_plug):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_DATA,
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == DEFAULT_NAME
assert result["data"] == CONF_DATA
async def test_flow_user_unknown_error(
hass: HomeAssistant, mocked_plug: MagicMock
) -> None:
"""Test user initialized flow with unreachable server."""
with patch_config_flow(mocked_plug) as mock:
mock.side_effect = Exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "unknown"
with patch_config_flow(mocked_plug):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_DATA,
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == DEFAULT_NAME
assert result["data"] == CONF_DATA
async def test_import(hass: HomeAssistant, mocked_plug: MagicMock) -> None:
"""Test import initialized flow."""
with patch_config_flow(mocked_plug), _patch_setup_entry():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=CONF_IMPORT_DATA,
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == "Smart Plug"
assert result["data"] == CONF_DATA