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

Clean up Insteon config_flow (#93004) 

* Clean up config_flow

* Remove unnecessary logging

* Remove logger

* Add more tests

* Test closing modem connection

* Simplify patching
pull/88559/head^2
Tom Harris 2023-05-23 11:31:17 -04:00 committed by GitHub
parent d62cdf3d65
commit e2fb1de0d9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 222 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3,8 +3,7 @@ from __future__ import annotations
import logging
from pyinsteon import async_connect
import voluptuous as vol
from pyinsteon import async_close, async_connect, devices
from homeassistant import config_entries
from homeassistant.components import dhcp, usb
@ -44,6 +43,7 @@ from .schemas import (
build_remove_x10_schema,
build_x10_schema,
)
from .utils import async_get_usb_ports
STEP_PLM = "plm"
STEP_HUB_V1 = "hubv1"
@ -55,18 +55,10 @@ STEP_ADD_OVERRIDE = "add_override"
STEP_REMOVE_OVERRIDE = "remove_override"
STEP_REMOVE_X10 = "remove_x10"
MODEM_TYPE = "modem_type"
PLM = "PowerLinc Modem (PLM)"
HUB1 = "Hub version 1 (pre-2014)"
HUB2 = "Hub version 2"
_LOGGER = logging.getLogger(__name__)
def _only_one_selected(*args):
"""Test if only one item is True."""
return sum(args) == 1
async def _async_connect(**kwargs):
"""Connect to the Insteon modem."""
try:
@ -128,22 +120,10 @@ class InsteonFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_user(self, user_input=None):
"""Init the config flow."""
errors = {}
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
if user_input is not None:
selection = user_input.get(MODEM_TYPE)
if selection == PLM:
return await self.async_step_plm()
if selection == HUB1:
return await self.async_step_hubv1()
return await self.async_step_hubv2()
modem_types = [PLM, HUB1, HUB2]
data_schema = vol.Schema({vol.Required(MODEM_TYPE): vol.In(modem_types)})
return self.async_show_form(
step_id="user", data_schema=data_schema, errors=errors
)
modem_types = [STEP_PLM, STEP_HUB_V1, STEP_HUB_V2]
return self.async_show_menu(step_id="user", menu_options=modem_types)
async def async_step_plm(self, user_input=None):
"""Set up the PLM modem type."""
@ -153,7 +133,8 @@ class InsteonFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title="", data=user_input)
errors["base"] = "cannot_connect"
schema_defaults = user_input if user_input is not None else {}
data_schema = build_plm_schema(**schema_defaults)
ports = await async_get_usb_ports(self.hass)
data_schema = build_plm_schema(ports, **schema_defaults)
return self.async_show_form(
step_id=STEP_PLM, data_schema=data_schema, errors=errors
)
@ -243,57 +224,24 @@ class InsteonOptionsFlowHandler(config_entries.OptionsFlow):
async def async_step_init(self, user_input=None) -> FlowResult:
"""Init the options config flow."""
errors = {}
if user_input is not None:
change_hub_config = user_input.get(STEP_CHANGE_HUB_CONFIG, False)
change_plm_config = user_input.get(STEP_CHANGE_PLM_CONFIG, False)
device_override = user_input.get(STEP_ADD_OVERRIDE, False)
x10_device = user_input.get(STEP_ADD_X10, False)
remove_override = user_input.get(STEP_REMOVE_OVERRIDE, False)
remove_x10 = user_input.get(STEP_REMOVE_X10, False)
if _only_one_selected(
change_hub_config,
change_plm_config,
device_override,
x10_device,
remove_override,
remove_x10,
):
if change_hub_config:
return await self.async_step_change_hub_config()
if change_plm_config:
return await self.async_step_change_plm_config()
if device_override:
return await self.async_step_add_override()
if x10_device:
return await self.async_step_add_x10()
if remove_override:
return await self.async_step_remove_override()
if remove_x10:
return await self.async_step_remove_x10()
errors["base"] = "select_single"
menu_options = [STEP_ADD_OVERRIDE, STEP_ADD_X10]
data_schema = {
vol.Optional(STEP_ADD_OVERRIDE): bool,
vol.Optional(STEP_ADD_X10): bool,
}
if self.config_entry.data.get(CONF_HOST):
data_schema[vol.Optional(STEP_CHANGE_HUB_CONFIG)] = bool
menu_options.append(STEP_CHANGE_HUB_CONFIG)
else:
data_schema[vol.Optional(STEP_CHANGE_PLM_CONFIG)] = bool
menu_options.append(STEP_CHANGE_PLM_CONFIG)
options = {**self.config_entry.options}
if options.get(CONF_OVERRIDE):
data_schema[vol.Optional(STEP_REMOVE_OVERRIDE)] = bool
menu_options.append(STEP_REMOVE_OVERRIDE)
if options.get(CONF_X10):
data_schema[vol.Optional(STEP_REMOVE_X10)] = bool
menu_options.append(STEP_REMOVE_X10)
return self.async_show_form(
step_id="init", data_schema=vol.Schema(data_schema), errors=errors
)
return self.async_show_menu(step_id="init", menu_options=menu_options)
async def async_step_change_hub_config(self, user_input=None) -> FlowResult:
"""Change the Hub configuration."""
errors = {}
if user_input is not None:
data = {
**self.config_entry.data,
@ -303,31 +251,41 @@ class InsteonOptionsFlowHandler(config_entries.OptionsFlow):
if self.config_entry.data[CONF_HUB_VERSION] == 2:
data[CONF_USERNAME] = user_input[CONF_USERNAME]
data[CONF_PASSWORD] = user_input[CONF_PASSWORD]
self.hass.config_entries.async_update_entry(self.config_entry, data=data)
return self.async_create_entry(
title="",
data={**self.config_entry.options},
)
if devices.modem:
await async_close()
if await _async_connect(**data):
self.hass.config_entries.async_update_entry(
self.config_entry, data=data
)
return self.async_create_entry(data={**self.config_entry.options})
errors["base"] = "cannot_connect"
data_schema = build_hub_schema(**self.config_entry.data)
return self.async_show_form(
step_id=STEP_CHANGE_HUB_CONFIG, data_schema=data_schema
step_id=STEP_CHANGE_HUB_CONFIG, data_schema=data_schema, errors=errors
)
async def async_step_change_plm_config(self, user_input=None) -> FlowResult:
"""Change the PLM configuration."""
errors = {}
if user_input is not None:
data = {
**self.config_entry.data,
CONF_DEVICE: user_input[CONF_DEVICE],
}
self.hass.config_entries.async_update_entry(self.config_entry, data=data)
return self.async_create_entry(
title="",
data={**self.config_entry.options},
)
data_schema = build_plm_schema(**self.config_entry.data)
if devices.modem:
await async_close()
if await _async_connect(**data):
self.hass.config_entries.async_update_entry(
self.config_entry, data=data
)
return self.async_create_entry(data={**self.config_entry.options})
errors["base"] = "cannot_connect"
ports = await async_get_usb_ports(self.hass)
data_schema = build_plm_schema(ports, **self.config_entry.data)
return self.async_show_form(
step_id=STEP_CHANGE_PLM_CONFIG, data_schema=data_schema
step_id=STEP_CHANGE_PLM_CONFIG, data_schema=data_schema, errors=errors
)
async def async_step_add_override(self, user_input=None) -> FlowResult:
@ -337,7 +295,7 @@ class InsteonOptionsFlowHandler(config_entries.OptionsFlow):
try:
data = add_device_override({**self.config_entry.options}, user_input)
async_dispatcher_send(self.hass, SIGNAL_ADD_DEVICE_OVERRIDE, user_input)
return self.async_create_entry(title="", data=data)
return self.async_create_entry(data=data)
except ValueError:
errors["base"] = "input_error"
schema_defaults = user_input if user_input is not None else {}
@ -352,7 +310,7 @@ class InsteonOptionsFlowHandler(config_entries.OptionsFlow):
if user_input is not None:
options = add_x10_device({**self.config_entry.options}, user_input)
async_dispatcher_send(self.hass, SIGNAL_ADD_X10_DEVICE, user_input)
return self.async_create_entry(title="", data=options)
return self.async_create_entry(data=options)
schema_defaults: dict[str, str] = user_input if user_input is not None else {}
data_schema = build_x10_schema(**schema_defaults)
return self.async_show_form(
@ -370,7 +328,7 @@ class InsteonOptionsFlowHandler(config_entries.OptionsFlow):
SIGNAL_REMOVE_DEVICE_OVERRIDE,
user_input[CONF_ADDRESS],
)
return self.async_create_entry(title="", data=options)
return self.async_create_entry(data=options)
data_schema = build_remove_override_schema(options[CONF_OVERRIDE])
return self.async_show_form(
@ -386,7 +344,7 @@ class InsteonOptionsFlowHandler(config_entries.OptionsFlow):
async_dispatcher_send(
self.hass, SIGNAL_REMOVE_X10_DEVICE, housecode, unitcode
)
return self.async_create_entry(title="", data=options)
return self.async_create_entry(data=options)
data_schema = build_remove_x10_schema(options[CONF_X10])
return self.async_show_form(

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

@ -285,9 +285,21 @@ def build_x10_schema(
)
def build_plm_schema(device=vol.UNDEFINED):
def _find_likely_port(ports):
"""Return the most likely USB port for a PLM."""
test_strings = ["FTDI", "0403:6001", "10BF:"]
for port, name in ports.items():
for test_string in test_strings:
if test_string in name:
return port
return vol.UNDEFINED
def build_plm_schema(ports: dict[str, str], device=vol.UNDEFINED):
"""Build the PLM schema for config flow."""
return vol.Schema({vol.Required(CONF_DEVICE, default=device): str})
if not device or device == vol.UNDEFINED:
device = _find_likely_port(ports)
return vol.Schema({vol.Required(CONF_DEVICE, default=device): vol.In(ports)})
def build_hub_schema(

16
homeassistant/components/insteon/strings.json
View File

@ -4,8 +4,10 @@
"step": {
"user": {
"description": "Select the Insteon modem type.",
"data": {
"modem_type": "Modem type."
"menu_options": {
"plm": "PowerLink Modem (PLM)",
"hubv1": "Hub version 1 (pre 2014)",
"hubv2": "Hub version 2"
}
},
"plm": {
@ -38,8 +40,7 @@
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"select_single": "Select one option."
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"abort": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
@ -50,7 +51,7 @@
"options": {
"step": {
"init": {
"data": {
"menu_options": {
"change_hub_config": "Change the Hub configuration.",
"change_plm_config": "Change the PLM configuration.",
"add_override": "Add a device override.",
@ -60,7 +61,7 @@
}
},
"change_hub_config": {
"description": "Change the Insteon Hub connection information. You must restart Home Assistant after making this change. This does not change the configuration of the Hub itself. To change the configuration in the Hub use the Hub app.",
"description": "Change the Insteon Hub connection information.",
"data": {
"host": "[%key:common::config_flow::data::ip%]",
"port": "[%key:common::config_flow::data::port%]",
@ -69,7 +70,7 @@
}
},
"change_plm_config": {
"description": "Change the Insteon PLM connection information. You must restart Home Assistant after making this change. This does not change the configuration of the PLM itself.",
"description": "Change the Insteon PLM connection information.",
"data": {
"device": "[%key:common::config_flow::data::usb_path%]"
}
@ -106,7 +107,6 @@
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"select_single": "Select one option.",
"input_error": "Invalid entries, please check your values."
}
}

31
homeassistant/components/insteon/utils.py
View File

@ -22,7 +22,9 @@ from pyinsteon.managers.x10_manager import (
async_x10_all_units_off,
)
from pyinsteon.x10_address import create as create_x10_address
from serial.tools import list_ports
from homeassistant.components import usb
from homeassistant.const import (
CONF_ADDRESS,
CONF_ENTITY_ID,
@ -393,3 +395,32 @@ def async_add_insteon_entities(
for group in groups:
new_entities.append(entity_type(device, group))
async_add_entities(new_entities)
def get_usb_ports() -> dict[str, str]:
"""Return a dict of USB ports and their friendly names."""
ports = list_ports.comports()
port_descriptions = {}
for port in ports:
vid: str | None = None
pid: str | None = None
if port.vid is not None and port.pid is not None:
usb_device = usb.usb_device_from_port(port)
vid = usb_device.vid
pid = usb_device.pid
dev_path = usb.get_serial_by_id(port.device)
human_name = usb.human_readable_device_name(
dev_path,
port.serial_number,
port.manufacturer,
port.description,
vid,
pid,
)
port_descriptions[dev_path] = human_name
return port_descriptions
async def async_get_usb_ports(hass: HomeAssistant) -> dict[str, str]:
"""Return a dict of USB ports and their friendly names."""
return await hass.async_add_executor_job(get_usb_ports)

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

@ -96,5 +96,8 @@ 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"
PATCH_USB_LIST = "homeassistant.components.insteon.config_flow.async_get_usb_ports"
PATCH_ASYNC_SETUP = "homeassistant.components.insteon.async_setup"
PATCH_ASYNC_SETUP_ENTRY = "homeassistant.components.insteon.async_setup_entry"

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

@ -2,18 +2,19 @@
from unittest.mock import patch
import pytest
from voluptuous_serialize import convert
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import usb
from homeassistant.components import dhcp, usb
from homeassistant.components.insteon.config_flow import (
HUB1,
HUB2,
MODEM_TYPE,
PLM,
STEP_ADD_OVERRIDE,
STEP_ADD_X10,
STEP_CHANGE_HUB_CONFIG,
STEP_CHANGE_PLM_CONFIG,
STEP_HUB_V1,
STEP_HUB_V2,
STEP_PLM,
STEP_REMOVE_OVERRIDE,
STEP_REMOVE_X10,
)
@ -40,6 +41,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from .const import (
MOCK_DEVICE,
MOCK_HOSTNAME,
MOCK_IMPORT_CONFIG_PLM,
MOCK_IMPORT_MINIMUM_HUB_V1,
@ -52,16 +54,37 @@ from .const import (
PATCH_ASYNC_SETUP,
PATCH_ASYNC_SETUP_ENTRY,
PATCH_CONNECTION,
PATCH_CONNECTION_CLOSE,
PATCH_DEVICES,
PATCH_USB_LIST,
)
from .mock_devices import MockDevices
from tests.common import MockConfigEntry
USB_PORTS = {"/dev/ttyUSB0": "/dev/ttyUSB0", MOCK_DEVICE: MOCK_DEVICE}
async def mock_successful_connection(*args, **kwargs):
"""Return a successful connection."""
return True
async def mock_usb_list(hass: HomeAssistant):
"""Return a mock list of USB devices."""
return USB_PORTS
@pytest.fixture(autouse=True)
def patch_usb_list():
"""Only setup the lock and required base platforms to speed up tests."""
with patch(
PATCH_USB_LIST,
mock_usb_list,
):
yield
async def mock_failed_connection(*args, **kwargs):
"""Return a failed connection."""
raise ConnectionError("Connection failed")
@ -72,12 +95,11 @@ async def _init_form(hass, modem_type):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {}
assert result["type"] == data_entry_flow.FlowResultType.MENU
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{MODEM_TYPE: modem_type},
{"next_step_id": modem_type},
)
return result2
@ -99,7 +121,7 @@ async def _device_form(hass, flow_id, connection, user_input):
async def test_form_select_modem(hass: HomeAssistant) -> None:
"""Test we get a modem form."""
result = await _init_form(hass, HUB2)
result = await _init_form(hass, STEP_HUB_V2)
assert result["step_id"] == STEP_HUB_V2
assert result["type"] == "form"
@ -127,7 +149,7 @@ async def test_fail_on_existing(hass: HomeAssistant) -> None:
async def test_form_select_plm(hass: HomeAssistant) -> None:
"""Test we set up the PLM correctly."""
result = await _init_form(hass, PLM)
result = await _init_form(hass, STEP_PLM)
result2, mock_setup, mock_setup_entry = await _device_form(
hass, result["flow_id"], mock_successful_connection, MOCK_USER_INPUT_PLM
@ -142,7 +164,7 @@ async def test_form_select_plm(hass: HomeAssistant) -> None:
async def test_form_select_hub_v1(hass: HomeAssistant) -> None:
"""Test we set up the Hub v1 correctly."""
result = await _init_form(hass, HUB1)
result = await _init_form(hass, STEP_HUB_V1)
result2, mock_setup, mock_setup_entry = await _device_form(
hass, result["flow_id"], mock_successful_connection, MOCK_USER_INPUT_HUB_V1
@ -160,7 +182,7 @@ async def test_form_select_hub_v1(hass: HomeAssistant) -> None:
async def test_form_select_hub_v2(hass: HomeAssistant) -> None:
"""Test we set up the Hub v2 correctly."""
result = await _init_form(hass, HUB2)
result = await _init_form(hass, STEP_HUB_V2)
result2, mock_setup, mock_setup_entry = await _device_form(
hass, result["flow_id"], mock_successful_connection, MOCK_USER_INPUT_HUB_V2
@ -175,10 +197,32 @@ async def test_form_select_hub_v2(hass: HomeAssistant) -> None:
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_discovery_dhcp(hass: HomeAssistant) -> None:
"""Test the discovery of the Hub via DHCP."""
discovery_info = dhcp.DhcpServiceInfo("1.2.3.4", "", "aa:bb:cc:dd:ee:ff")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=discovery_info
)
assert result["type"] == data_entry_flow.FlowResultType.MENU
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"next_step_id": STEP_HUB_V2},
)
assert result2["type"] == data_entry_flow.FlowResultType.FORM
schema = convert(result2["data_schema"])
found_host = False
for field in schema:
if field["name"] == CONF_HOST:
assert field["default"] == "1.2.3.4"
found_host = True
assert found_host
async def test_failed_connection_plm(hass: HomeAssistant) -> None:
"""Test a failed connection with the PLM."""
result = await _init_form(hass, PLM)
result = await _init_form(hass, STEP_PLM)
result2, _, _ = await _device_form(
hass, result["flow_id"], mock_failed_connection, MOCK_USER_INPUT_PLM
@ -190,7 +234,7 @@ async def test_failed_connection_plm(hass: HomeAssistant) -> None:
async def test_failed_connection_hub(hass: HomeAssistant) -> None:
"""Test a failed connection with a Hub."""
result = await _init_form(hass, HUB2)
result = await _init_form(hass, STEP_HUB_V2)
result2, _, _ = await _device_form(
hass, result["flow_id"], mock_failed_connection, MOCK_USER_INPUT_HUB_V2
@ -228,12 +272,12 @@ async def _options_init_form(hass, entry_id, step):
with patch(PATCH_ASYNC_SETUP_ENTRY, return_value=True):
result = await hass.config_entries.options.async_init(entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["type"] == data_entry_flow.FlowResultType.MENU
assert result["step_id"] == "init"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
{step: True},
{"next_step_id": step},
)
return result2
@ -307,10 +351,18 @@ async def test_import_failed_connection(hass: HomeAssistant) -> None:
assert result["reason"] == "cannot_connect"
async def _options_form(hass, flow_id, user_input):
async def _options_form(
hass, flow_id, user_input, connection=mock_successful_connection
):
"""Test an options form."""
with patch(PATCH_ASYNC_SETUP_ENTRY, return_value=True) as mock_setup_entry:
mock_devices = MockDevices(connected=True)
await mock_devices.async_load()
mock_devices.modem = mock_devices["AA.AA.AA"]
with patch(PATCH_CONNECTION, new=connection), patch(
PATCH_ASYNC_SETUP_ENTRY, return_value=True
) as mock_setup_entry, patch(PATCH_DEVICES, mock_devices), patch(
PATCH_CONNECTION_CLOSE
):
result = await hass.config_entries.options.async_configure(flow_id, user_input)
return result, mock_setup_entry
@ -336,12 +388,39 @@ async def test_options_change_hub_config(hass: HomeAssistant) -> None:
CONF_PASSWORD: "new password",
}
result, _ = await _options_form(hass, result["flow_id"], user_input)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert config_entry.options == {}
assert config_entry.data == {**user_input, CONF_HUB_VERSION: 2}
async def test_options_change_hub_bad_config(hass: HomeAssistant) -> None:
"""Test changing Hub v2 with bad config."""
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)
result = await _options_init_form(
hass, config_entry.entry_id, STEP_CHANGE_HUB_CONFIG
)
user_input = {
CONF_HOST: "2.3.4.5",
CONF_PORT: 9999,
CONF_USERNAME: "new username",
CONF_PASSWORD: "new password",
}
result, _ = await _options_form(
hass, result["flow_id"], user_input, mock_failed_connection
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"]["base"] == "cannot_connect"
async def test_options_change_plm_config(hass: HomeAssistant) -> None:
"""Test changing PLM config."""
config_entry = MockConfigEntry(
@ -356,7 +435,7 @@ async def test_options_change_plm_config(hass: HomeAssistant) -> None:
hass, config_entry.entry_id, STEP_CHANGE_PLM_CONFIG
)
user_input = {CONF_DEVICE: "/dev/some_other_device"}
user_input = {CONF_DEVICE: "/dev/ttyUSB0"}
result, _ = await _options_form(hass, result["flow_id"], user_input)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
@ -364,6 +443,31 @@ async def test_options_change_plm_config(hass: HomeAssistant) -> None:
assert config_entry.data == user_input
async def test_options_change_plm_bad_config(hass: HomeAssistant) -> None:
"""Test changing PLM config."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data=MOCK_USER_INPUT_PLM,
options={},
)
config_entry.add_to_hass(hass)
result = await _options_init_form(
hass, config_entry.entry_id, STEP_CHANGE_PLM_CONFIG
)
user_input = {CONF_DEVICE: "/dev/ttyUSB0"}
result, _ = await _options_form(
hass, result["flow_id"], user_input, mock_failed_connection
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"]["base"] == "cannot_connect"
async def test_options_add_device_override(hass: HomeAssistant) -> None:
"""Test adding a device override."""
config_entry = MockConfigEntry(
@ -581,28 +685,6 @@ async def test_options_remove_x10_device_with_override(hass: HomeAssistant) -> N
assert len(config_entry.options[CONF_OVERRIDE]) == 1
async def test_options_dup_selection(hass: HomeAssistant) -> None:
"""Test if a duplicate selection was made in options."""
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)
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
{STEP_ADD_OVERRIDE: True, STEP_ADD_X10: True},
)
assert result2["type"] == data_entry_flow.FlowResultType.FORM
assert result2["errors"] == {"base": "select_single"}
async def test_options_override_bad_data(hass: HomeAssistant) -> None:
"""Test for bad data in a device override."""
@ -644,9 +726,7 @@ async def test_discovery_via_usb(hass: HomeAssistant) -> None:
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "confirm_usb"
with patch("homeassistant.components.insteon.config_flow.async_connect"), patch(
"homeassistant.components.insteon.async_setup_entry", return_value=True
):
with patch(PATCH_CONNECTION), patch(PATCH_ASYNC_SETUP, return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)