jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-28 07:37:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove insteon import flow (#93952)

Browse Source
This commit is contained in:
Erik Montnemery 2023-06-02 09:21:19 +02:00 committed by GitHub
parent f941203949
commit 0916a6c0cb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 6 additions and 455 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
homeassistant/components/insteon/__init__.py
View File

@ -9,13 +9,12 @@ from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_PLATFORM, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.typing import ConfigType
from . import api
from .const import (
CONF_CAT,
CONF_DEV_PATH,
CONF_DIM_STEPS,
CONF_HOUSECODE,
CONF_OVERRIDE,
@ -25,7 +24,6 @@ from .const import (
DOMAIN,
INSTEON_PLATFORMS,
)
from .schemas import convert_yaml_to_config_flow
from .utils import (
add_insteon_events,
async_register_services,
@ -36,6 +34,8 @@ from .utils import (
_LOGGER = logging.getLogger(__name__)
OPTIONS = "options"
CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
async def async_get_device_config(hass, config_entry):
"""Initiate the connection and services."""
@ -77,26 +77,6 @@ async def close_insteon_connection(*args):
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Insteon platform."""
hass.data[DOMAIN] = {}
if DOMAIN not in config:
return True
conf = dict(config[DOMAIN])
hass.data[DOMAIN][CONF_DEV_PATH] = conf.pop(CONF_DEV_PATH, None)
if not conf:
return True
data, options = convert_yaml_to_config_flow(conf)
if options:
hass.data[DOMAIN][OPTIONS] = options
# Create a config entry with the connection data
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=data
)
)
return True

8
homeassistant/components/insteon/config_flow.py
View File

@ -163,14 +163,6 @@ class InsteonFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
step_id=step_id, data_schema=data_schema, errors=errors
)
async def async_step_import(self, import_info):
"""Import a yaml entry as a config entry."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
if not await _async_connect(**import_info):
return self.async_abort(reason="cannot_connect")
return self.async_create_entry(title="", data=import_info)
async def async_step_usb(self, discovery_info: usb.UsbServiceInfo) -> FlowResult:
"""Handle USB discovery."""
if self._async_current_entries():

136
homeassistant/components/insteon/schemas.py
View File

@ -22,26 +22,13 @@ import homeassistant.helpers.config_validation as cv
from .const import (
CONF_CAT,
CONF_DEV_PATH,
CONF_DIM_STEPS,
CONF_FIRMWARE,
CONF_HOUSECODE,
CONF_HUB_PASSWORD,
CONF_HUB_USERNAME,
CONF_HUB_VERSION,
CONF_IP_PORT,
CONF_OVERRIDE,
CONF_PLM_HUB_MSG,
CONF_PRODUCT_KEY,
CONF_SUBCAT,
CONF_UNITCODE,
CONF_X10,
CONF_X10_ALL_LIGHTS_OFF,
CONF_X10_ALL_LIGHTS_ON,
CONF_X10_ALL_UNITS_OFF,
DOMAIN,
HOUSECODES,
INSTEON_ADDR_REGEX,
PORT_HUB_V1,
PORT_HUB_V2,
SRV_ALL_LINK_GROUP,
@ -53,88 +40,6 @@ from .const import (
X10_PLATFORMS,
)
def set_default_port(schema: dict) -> dict:
"""Set the default port based on the Hub version."""
# If the ip_port is found do nothing
# If it is not found the set the default
if not schema.get(CONF_IP_PORT):
hub_version = schema.get(CONF_HUB_VERSION)
# Found hub_version but not ip_port
schema[CONF_IP_PORT] = PORT_HUB_V1 if hub_version == 1 else PORT_HUB_V2
return schema
def insteon_address(value: str) -> str:
"""Validate an Insteon address."""
if not INSTEON_ADDR_REGEX.match(value):
raise vol.Invalid("Invalid Insteon Address")
return str(value).replace(".", "").lower()
CONF_DEVICE_OVERRIDE_SCHEMA = vol.All(
vol.Schema(
{
vol.Required(CONF_ADDRESS): cv.string,
vol.Optional(CONF_CAT): cv.byte,
vol.Optional(CONF_SUBCAT): cv.byte,
vol.Optional(CONF_FIRMWARE): cv.byte,
vol.Optional(CONF_PRODUCT_KEY): cv.byte,
vol.Optional(CONF_PLATFORM): cv.string,
}
),
)
CONF_X10_SCHEMA = vol.All(
vol.Schema(
{
vol.Required(CONF_HOUSECODE): cv.string,
vol.Required(CONF_UNITCODE): vol.Range(min=1, max=16),
vol.Required(CONF_PLATFORM): cv.string,
vol.Optional(CONF_DIM_STEPS): vol.Range(min=2, max=255),
}
)
)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
cv.deprecated(CONF_X10_ALL_UNITS_OFF),
cv.deprecated(CONF_X10_ALL_LIGHTS_ON),
cv.deprecated(CONF_X10_ALL_LIGHTS_OFF),
vol.Schema(
{
vol.Exclusive(
CONF_PORT, "plm_or_hub", msg=CONF_PLM_HUB_MSG
): cv.string,
vol.Exclusive(
CONF_HOST, "plm_or_hub", msg=CONF_PLM_HUB_MSG
): cv.string,
vol.Optional(CONF_IP_PORT): cv.port,
vol.Optional(CONF_HUB_USERNAME): cv.string,
vol.Optional(CONF_HUB_PASSWORD): cv.string,
vol.Optional(CONF_HUB_VERSION, default=2): vol.In([1, 2]),
vol.Optional(CONF_OVERRIDE): vol.All(
cv.ensure_list_csv, [CONF_DEVICE_OVERRIDE_SCHEMA]
),
vol.Optional(CONF_X10): vol.All(
cv.ensure_list_csv, [CONF_X10_SCHEMA]
),
vol.Optional(CONF_DEV_PATH): cv.string,
},
extra=vol.ALLOW_EXTRA,
required=True,
),
cv.has_at_least_one_key(CONF_PORT, CONF_HOST),
set_default_port,
)
},
extra=vol.ALLOW_EXTRA,
)
ADD_ALL_LINK_SCHEMA = vol.Schema(
{
vol.Required(SRV_ALL_LINK_GROUP): vol.Range(min=0, max=255),
@ -170,18 +75,6 @@ TRIGGER_SCENE_SCHEMA = vol.Schema(
ADD_DEFAULT_LINKS_SCHEMA = vol.Schema({vol.Required(CONF_ENTITY_ID): cv.entity_id})
SCENE_ENTITY_SCHEMA = vol.Schema(
[
{
vol.Required(CONF_ADDRESS): str,
vol.Required("data1"): int,
vol.Required("data2"): int,
vol.Required("data3"): int,
}
]
)
def normalize_byte_entry_to_int(entry: int | bytes | str):
"""Format a hex entry value."""
if isinstance(entry, int):
@ -338,32 +231,3 @@ def build_remove_x10_schema(data):
unitcode = device[CONF_UNITCODE]
selection.append(f"Housecode: {housecode}, Unitcode: {unitcode}")
return vol.Schema({vol.Required(CONF_DEVICE): vol.In(selection)})
def convert_yaml_to_config_flow(yaml_config):
"""Convert the YAML based configuration to a config flow configuration."""
config = {}
if yaml_config.get(CONF_HOST):
hub_version = yaml_config.get(CONF_HUB_VERSION, 2)
default_port = PORT_HUB_V2 if hub_version == 2 else PORT_HUB_V1
config[CONF_HOST] = yaml_config.get(CONF_HOST)
config[CONF_PORT] = yaml_config.get(CONF_PORT, default_port)
config[CONF_HUB_VERSION] = hub_version
if hub_version == 2:
config[CONF_USERNAME] = yaml_config[CONF_USERNAME]
config[CONF_PASSWORD] = yaml_config[CONF_PASSWORD]
else:
config[CONF_DEVICE] = yaml_config[CONF_PORT]
options = {}
for old_override in yaml_config.get(CONF_OVERRIDE, []):
override = {}
override[CONF_ADDRESS] = str(Address(old_override[CONF_ADDRESS]))
override[CONF_CAT] = normalize_byte_entry_to_int(old_override[CONF_CAT])
override[CONF_SUBCAT] = normalize_byte_entry_to_int(old_override[CONF_SUBCAT])
options = add_device_override(options, override)
for x10_device in yaml_config.get(CONF_X10, []):
options = add_x10_device(options, x10_device)
return config, options

25
tests/components/insteon/const.py
View File

@ -3,11 +3,8 @@ from homeassistant.components.insteon.const import (
CONF_CAT,
CONF_DIM_STEPS,
CONF_HOUSECODE,
CONF_HUB_VERSION,
CONF_OVERRIDE,
CONF_SUBCAT,
CONF_UNITCODE,
CONF_X10,
X10_PLATFORMS,
)
from homeassistant.const import (
@ -73,28 +70,6 @@ MOCK_X10_CONFIG_2 = {
CONF_DIM_STEPS: MOCK_X10_STEPS,
}
MOCK_IMPORT_CONFIG_PLM = {CONF_PORT: MOCK_DEVICE}
MOCK_IMPORT_MINIMUM_HUB_V2 = {
CONF_HOST: MOCK_HOSTNAME,
CONF_USERNAME: MOCK_USERNAME,
CONF_PASSWORD: MOCK_PASSWORD,
}
MOCK_IMPORT_MINIMUM_HUB_V1 = {CONF_HOST: MOCK_HOSTNAME, CONF_HUB_VERSION: 1}
MOCK_IMPORT_FULL_CONFIG_PLM = MOCK_IMPORT_CONFIG_PLM.copy()
MOCK_IMPORT_FULL_CONFIG_PLM[CONF_OVERRIDE] = [MOCK_DEVICE_OVERRIDE_CONFIG]
MOCK_IMPORT_FULL_CONFIG_PLM[CONF_X10] = [MOCK_X10_CONFIG_1, MOCK_X10_CONFIG_2]
MOCK_IMPORT_FULL_CONFIG_HUB_V2 = MOCK_USER_INPUT_HUB_V2.copy()
MOCK_IMPORT_FULL_CONFIG_HUB_V2[CONF_HUB_VERSION] = 2
MOCK_IMPORT_FULL_CONFIG_HUB_V2[CONF_OVERRIDE] = [MOCK_DEVICE_OVERRIDE_CONFIG]
MOCK_IMPORT_FULL_CONFIG_HUB_V2[CONF_X10] = [MOCK_X10_CONFIG_1, MOCK_X10_CONFIG_2]
MOCK_IMPORT_FULL_CONFIG_HUB_V1 = MOCK_USER_INPUT_HUB_V1.copy()
MOCK_IMPORT_FULL_CONFIG_HUB_V1[CONF_HUB_VERSION] = 1
MOCK_IMPORT_FULL_CONFIG_HUB_V1[CONF_OVERRIDE] = [MOCK_DEVICE_OVERRIDE_CONFIG]
MOCK_IMPORT_FULL_CONFIG_HUB_V1[CONF_X10] = [MOCK_X10_CONFIG_1, MOCK_X10_CONFIG_2]
PATCH_CONNECTION = "homeassistant.components.insteon.config_flow.async_connect"
PATCH_CONNECTION_CLOSE = "homeassistant.components.insteon.config_flow.async_close"
PATCH_DEVICES = "homeassistant.components.insteon.config_flow.devices"

99
tests/components/insteon/test_config_flow.py
View File

@ -42,15 +42,9 @@ from homeassistant.core import HomeAssistant
from .const import (
MOCK_DEVICE,
MOCK_HOSTNAME,
MOCK_IMPORT_CONFIG_PLM,
MOCK_IMPORT_MINIMUM_HUB_V1,
MOCK_IMPORT_MINIMUM_HUB_V2,
MOCK_PASSWORD,
MOCK_USER_INPUT_HUB_V1,
MOCK_USER_INPUT_HUB_V2,
MOCK_USER_INPUT_PLM,
MOCK_USERNAME,
PATCH_ASYNC_SETUP,
PATCH_ASYNC_SETUP_ENTRY,
PATCH_CONNECTION,
@ -243,30 +237,6 @@ async def test_failed_connection_hub(hass: HomeAssistant) -> None:
assert result2["errors"] == {"base": "cannot_connect"}
async def _import_config(hass, config):
"""Run the import step."""
with patch(
PATCH_CONNECTION,
new=mock_successful_connection,
), patch(
PATCH_ASYNC_SETUP, return_value=True
), patch(PATCH_ASYNC_SETUP_ENTRY, return_value=True):
return await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
async def test_import_plm(hass: HomeAssistant) -> None:
"""Test importing a minimum PLM config from yaml."""
result = await _import_config(hass, MOCK_IMPORT_CONFIG_PLM)
assert result["type"] == "create_entry"
assert hass.config_entries.async_entries(DOMAIN)
for entry in hass.config_entries.async_entries(DOMAIN):
assert entry.data == MOCK_IMPORT_CONFIG_PLM
async def _options_init_form(hass, entry_id, step):
"""Run the init options form."""
with patch(PATCH_ASYNC_SETUP_ENTRY, return_value=True):
@ -282,75 +252,6 @@ async def _options_init_form(hass, entry_id, step):
return result2
async def test_import_min_hub_v2(hass: HomeAssistant) -> None:
"""Test importing a minimum Hub v2 config from yaml."""
result = await _import_config(
hass, {**MOCK_IMPORT_MINIMUM_HUB_V2, CONF_PORT: 25105, CONF_HUB_VERSION: 2}
)
assert result["type"] == "create_entry"
assert hass.config_entries.async_entries(DOMAIN)
for entry in hass.config_entries.async_entries(DOMAIN):
assert entry.data[CONF_HOST] == MOCK_HOSTNAME
assert entry.data[CONF_PORT] == 25105
assert entry.data[CONF_USERNAME] == MOCK_USERNAME
assert entry.data[CONF_PASSWORD] == MOCK_PASSWORD
assert entry.data[CONF_HUB_VERSION] == 2
async def test_import_min_hub_v1(hass: HomeAssistant) -> None:
"""Test importing a minimum Hub v1 config from yaml."""
result = await _import_config(
hass, {**MOCK_IMPORT_MINIMUM_HUB_V1, CONF_PORT: 9761, CONF_HUB_VERSION: 1}
)
assert result["type"] == "create_entry"
assert hass.config_entries.async_entries(DOMAIN)
for entry in hass.config_entries.async_entries(DOMAIN):
assert entry.data[CONF_HOST] == MOCK_HOSTNAME
assert entry.data[CONF_PORT] == 9761
assert entry.data[CONF_HUB_VERSION] == 1
async def test_import_existing(hass: HomeAssistant) -> None:
"""Test we fail on an existing config imported."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={},
)
config_entry.add_to_hass(hass)
assert config_entry.state is config_entries.ConfigEntryState.NOT_LOADED
result = await _import_config(
hass, {**MOCK_IMPORT_MINIMUM_HUB_V2, CONF_PORT: 25105, CONF_HUB_VERSION: 2}
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"
async def test_import_failed_connection(hass: HomeAssistant) -> None:
"""Test a failed connection on import."""
with patch(
PATCH_CONNECTION,
new=mock_failed_connection,
), patch(
PATCH_ASYNC_SETUP, return_value=True
), patch(PATCH_ASYNC_SETUP_ENTRY, return_value=True):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={**MOCK_IMPORT_MINIMUM_HUB_V2, CONF_PORT: 25105, CONF_HUB_VERSION: 2},
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
async def _options_form(
hass, flow_id, user_input, connection=mock_successful_connection
):

167
tests/components/insteon/test_init.py
View File

@ -2,45 +2,15 @@
import asyncio
from unittest.mock import patch
from pyinsteon.address import Address
import pytest
from homeassistant.components import insteon
from homeassistant.components.insteon.const import (
CONF_CAT,
CONF_DEV_PATH,
CONF_OVERRIDE,
CONF_SUBCAT,
CONF_X10,
DOMAIN,
PORT_HUB_V1,
PORT_HUB_V2,
)
from homeassistant.const import (
CONF_ADDRESS,
CONF_DEVICE,
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.components.insteon.const import CONF_DEV_PATH, DOMAIN
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .const import (
MOCK_ADDRESS,
MOCK_CAT,
MOCK_IMPORT_CONFIG_PLM,
MOCK_IMPORT_FULL_CONFIG_HUB_V1,
MOCK_IMPORT_FULL_CONFIG_HUB_V2,
MOCK_IMPORT_FULL_CONFIG_PLM,
MOCK_IMPORT_MINIMUM_HUB_V1,
MOCK_IMPORT_MINIMUM_HUB_V2,
MOCK_SUBCAT,
MOCK_USER_INPUT_PLM,
PATCH_CONNECTION,
)
from .const import MOCK_USER_INPUT_PLM, PATCH_CONNECTION
from .mock_devices import MockDevices
from tests.common import MockConfigEntry
@ -79,137 +49,6 @@ async def test_setup_entry(hass: HomeAssistant) -> None:
assert mock_close.called
async def test_import_plm(hass: HomeAssistant) -> None:
"""Test setting up the entry from YAML to a PLM."""
config = {}
config[DOMAIN] = MOCK_IMPORT_CONFIG_PLM
with patch.object(
insteon, "async_connect", new=mock_successful_connection
), patch.object(insteon, "close_insteon_connection"), patch.object(
insteon, "devices", new=MockDevices()
), patch(
PATCH_CONNECTION, new=mock_successful_connection
):
assert await async_setup_component(
hass,
insteon.DOMAIN,
config,
)
await hass.async_block_till_done()
await asyncio.sleep(0.01)
assert hass.config_entries.async_entries(DOMAIN)
data = hass.config_entries.async_entries(DOMAIN)[0].data
assert data[CONF_DEVICE] == MOCK_IMPORT_CONFIG_PLM[CONF_PORT]
assert CONF_PORT not in data
async def test_import_hub1(hass: HomeAssistant) -> None:
"""Test setting up the entry from YAML to a hub v1."""
config = {}
config[DOMAIN] = MOCK_IMPORT_MINIMUM_HUB_V1
with patch.object(
insteon, "async_connect", new=mock_successful_connection
), patch.object(insteon, "close_insteon_connection"), patch.object(
insteon, "devices", new=MockDevices()
), patch(
PATCH_CONNECTION, new=mock_successful_connection
):
assert await async_setup_component(
hass,
insteon.DOMAIN,
config,
)
await hass.async_block_till_done()
await asyncio.sleep(0.01)
assert hass.config_entries.async_entries(DOMAIN)
data = hass.config_entries.async_entries(DOMAIN)[0].data
assert data[CONF_HOST] == MOCK_IMPORT_FULL_CONFIG_HUB_V1[CONF_HOST]
assert data[CONF_PORT] == PORT_HUB_V1
assert CONF_USERNAME not in data
assert CONF_PASSWORD not in data
async def test_import_hub2(hass: HomeAssistant) -> None:
"""Test setting up the entry from YAML to a hub v2."""
config = {}
config[DOMAIN] = MOCK_IMPORT_MINIMUM_HUB_V2
with patch.object(
insteon, "async_connect", new=mock_successful_connection
), patch.object(insteon, "close_insteon_connection"), patch.object(
insteon, "devices", new=MockDevices()
), patch(
PATCH_CONNECTION, new=mock_successful_connection
):
assert await async_setup_component(
hass,
insteon.DOMAIN,
config,
)
await hass.async_block_till_done()
await asyncio.sleep(0.01)
assert hass.config_entries.async_entries(DOMAIN)
data = hass.config_entries.async_entries(DOMAIN)[0].data
assert data[CONF_HOST] == MOCK_IMPORT_FULL_CONFIG_HUB_V2[CONF_HOST]
assert data[CONF_PORT] == PORT_HUB_V2
assert data[CONF_USERNAME] == MOCK_IMPORT_MINIMUM_HUB_V2[CONF_USERNAME]
assert data[CONF_PASSWORD] == MOCK_IMPORT_MINIMUM_HUB_V2[CONF_PASSWORD]
async def test_import_options(hass: HomeAssistant) -> None:
"""Test setting up the entry from YAML including options."""
config = {}
config[DOMAIN] = MOCK_IMPORT_FULL_CONFIG_PLM
with patch.object(
insteon, "async_connect", new=mock_successful_connection
), patch.object(insteon, "close_insteon_connection"), patch.object(
insteon, "devices", new=MockDevices()
), patch(
PATCH_CONNECTION, new=mock_successful_connection
):
assert await async_setup_component(
hass,
insteon.DOMAIN,
config,
)
await hass.async_block_till_done()
await asyncio.sleep(0.01) # Need to yield to async processes
# pylint: disable=no-member
assert insteon.devices.add_x10_device.call_count == 2
assert insteon.devices.set_id.call_count == 1
options = hass.config_entries.async_entries(DOMAIN)[0].options
assert len(options[CONF_OVERRIDE]) == 1
assert options[CONF_OVERRIDE][0][CONF_ADDRESS] == str(Address(MOCK_ADDRESS))
assert options[CONF_OVERRIDE][0][CONF_CAT] == MOCK_CAT
assert options[CONF_OVERRIDE][0][CONF_SUBCAT] == MOCK_SUBCAT
assert len(options[CONF_X10]) == 2
assert options[CONF_X10][0] == MOCK_IMPORT_FULL_CONFIG_PLM[CONF_X10][0]
assert options[CONF_X10][1] == MOCK_IMPORT_FULL_CONFIG_PLM[CONF_X10][1]
async def test_import_failed_connection(hass: HomeAssistant) -> None:
"""Test a failed connection in import does not create a config entry."""
config = {}
config[DOMAIN] = MOCK_IMPORT_CONFIG_PLM
with patch.object(
insteon, "async_connect", new=mock_failed_connection
), patch.object(insteon, "async_close"), patch.object(
insteon, "devices", new=MockDevices(connected=False)
):
assert await async_setup_component(
hass,
insteon.DOMAIN,
config,
)
await hass.async_block_till_done()
assert not hass.config_entries.async_entries(DOMAIN)
async def test_setup_entry_failed_connection(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None: