diff --git a/homeassistant/components/insteon/__init__.py b/homeassistant/components/insteon/__init__.py
index 529ac20df5259841e7e29d02dc6329df6cb982ab..0ec2434bc823bfc9413f06fc3352128a14964bde 100644
--- a/homeassistant/components/insteon/__init__.py
+++ b/homeassistant/components/insteon/__init__.py
@@ -16,6 +16,7 @@ from homeassistant.helpers.typing import ConfigType
from . import api
from .const import (
CONF_CAT,
+ CONF_DEV_PATH,
CONF_DIM_STEPS,
CONF_HOUSECODE,
CONF_OVERRIDE,
@@ -84,6 +85,13 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up an Insteon entry."""
+ if dev_path := entry.options.get(CONF_DEV_PATH):
+ hass.data[DOMAIN] = {}
+ hass.data[DOMAIN][CONF_DEV_PATH] = dev_path
+
+ api.async_load_api(hass)
+ await api.async_register_insteon_frontend(hass)
+
if not devices.modem:
try:
await async_connect(**entry.data)
@@ -149,9 +157,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
create_insteon_device(hass, devices.modem, entry.entry_id)
- api.async_load_api(hass)
- await api.async_register_insteon_frontend(hass)
-
entry.async_create_background_task(
hass, async_get_device_config(hass, entry), "insteon-get-device-config"
)
diff --git a/homeassistant/components/insteon/api/__init__.py b/homeassistant/components/insteon/api/__init__.py
index fa006c6a6d98c790292fceeba0051b30f60765e4..1f671aa1343209bb7b2dcac3e77ed4cee56116a1 100644
--- a/homeassistant/components/insteon/api/__init__.py
+++ b/homeassistant/components/insteon/api/__init__.py
@@ -16,10 +16,19 @@ from .aldb import (
websocket_reset_aldb,
websocket_write_aldb,
)
+from .config import (
+ websocket_add_device_override,
+ websocket_get_config,
+ websocket_get_modem_schema,
+ websocket_remove_device_override,
+ websocket_update_modem_config,
+)
from .device import (
websocket_add_device,
+ websocket_add_x10_device,
websocket_cancel_add_device,
websocket_get_device,
+ websocket_remove_device,
)
from .properties import (
websocket_change_properties_record,
@@ -58,6 +67,8 @@ def async_load_api(hass):
websocket_api.async_register_command(hass, websocket_reset_aldb)
websocket_api.async_register_command(hass, websocket_add_default_links)
websocket_api.async_register_command(hass, websocket_notify_on_aldb_status)
+ websocket_api.async_register_command(hass, websocket_add_x10_device)
+ websocket_api.async_register_command(hass, websocket_remove_device)
websocket_api.async_register_command(hass, websocket_get_properties)
websocket_api.async_register_command(hass, websocket_change_properties_record)
@@ -65,6 +76,12 @@ def async_load_api(hass):
websocket_api.async_register_command(hass, websocket_load_properties)
websocket_api.async_register_command(hass, websocket_reset_properties)
+ websocket_api.async_register_command(hass, websocket_get_config)
+ websocket_api.async_register_command(hass, websocket_get_modem_schema)
+ websocket_api.async_register_command(hass, websocket_update_modem_config)
+ websocket_api.async_register_command(hass, websocket_add_device_override)
+ websocket_api.async_register_command(hass, websocket_remove_device_override)
+
async def async_register_insteon_frontend(hass: HomeAssistant):
"""Register the Insteon frontend configuration panel."""
@@ -80,8 +97,7 @@ async def async_register_insteon_frontend(hass: HomeAssistant):
hass=hass,
frontend_url_path=DOMAIN,
webcomponent_name="insteon-frontend",
- sidebar_title=DOMAIN.capitalize(),
- sidebar_icon="mdi:power",
+ config_panel_domain=DOMAIN,
module_url=f"{URL_BASE}/entrypoint-{build_id}.js",
embed_iframe=True,
require_admin=True,
diff --git a/homeassistant/components/insteon/api/config.py b/homeassistant/components/insteon/api/config.py
new file mode 100644
index 0000000000000000000000000000000000000000..8a617911d1ec069bc4b2870b86e6a75fb3a35727
--- /dev/null
+++ b/homeassistant/components/insteon/api/config.py
@@ -0,0 +1,272 @@
+"""API calls to manage Insteon configuration changes."""
+
+from __future__ import annotations
+
+from typing import Any, TypedDict
+
+from pyinsteon import async_close, async_connect, devices
+from pyinsteon.address import Address
+import voluptuous as vol
+import voluptuous_serialize
+
+from homeassistant.components import websocket_api
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_ADDRESS, CONF_DEVICE
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.dispatcher import async_dispatcher_send
+
+from ..const import (
+ CONF_HOUSECODE,
+ CONF_OVERRIDE,
+ CONF_UNITCODE,
+ CONF_X10,
+ DEVICE_ADDRESS,
+ DOMAIN,
+ ID,
+ SIGNAL_ADD_DEVICE_OVERRIDE,
+ SIGNAL_ADD_X10_DEVICE,
+ SIGNAL_REMOVE_DEVICE_OVERRIDE,
+ TYPE,
+)
+from ..schemas import (
+ build_device_override_schema,
+ build_hub_schema,
+ build_plm_manual_schema,
+ build_plm_schema,
+)
+from ..utils import async_get_usb_ports
+
+HUB_V1_SCHEMA = build_hub_schema(hub_version=1)
+HUB_V2_SCHEMA = build_hub_schema(hub_version=2)
+PLM_SCHEMA = build_plm_manual_schema()
+DEVICE_OVERRIDE_SCHEMA = build_device_override_schema()
+OVERRIDE = "override"
+
+
+class X10DeviceConfig(TypedDict):
+ """X10 Device Configuration Definition."""
+
+ housecode: str
+ unitcode: int
+ platform: str
+ dim_steps: int
+
+
+class DeviceOverride(TypedDict):
+ """X10 Device Configuration Definition."""
+
+ address: Address | str
+ cat: int
+ subcat: str
+
+
+def get_insteon_config_entry(hass: HomeAssistant) -> ConfigEntry:
+ """Return the Insteon configuration entry."""
+ return hass.config_entries.async_entries(DOMAIN)[0]
+
+
+def add_x10_device(hass: HomeAssistant, x10_device: X10DeviceConfig):
+ """Add an X10 device to the Insteon integration."""
+
+ config_entry = get_insteon_config_entry(hass)
+ x10_config = config_entry.options.get(CONF_X10, [])
+ if any(
+ device[CONF_HOUSECODE] == x10_device["housecode"]
+ and device[CONF_UNITCODE] == x10_device["unitcode"]
+ for device in x10_config
+ ):
+ raise ValueError("Duplicate X10 device")
+
+ hass.config_entries.async_update_entry(
+ entry=config_entry,
+ options=config_entry.options | {CONF_X10: [*x10_config, x10_device]},
+ )
+ async_dispatcher_send(hass, SIGNAL_ADD_X10_DEVICE, x10_device)
+
+
+def remove_x10_device(hass: HomeAssistant, housecode: str, unitcode: int):
+ """Remove an X10 device from the config."""
+
+ config_entry = get_insteon_config_entry(hass)
+ new_options = {**config_entry.options}
+ new_x10 = [
+ existing_device
+ for existing_device in config_entry.options.get(CONF_X10, [])
+ if existing_device[CONF_HOUSECODE].lower() != housecode.lower()
+ or existing_device[CONF_UNITCODE] != unitcode
+ ]
+
+ new_options[CONF_X10] = new_x10
+ hass.config_entries.async_update_entry(entry=config_entry, options=new_options)
+
+
+def add_device_overide(hass: HomeAssistant, override: DeviceOverride):
+ """Add an Insteon device override."""
+
+ config_entry = get_insteon_config_entry(hass)
+ override_config = config_entry.options.get(CONF_OVERRIDE, [])
+ address = Address(override[CONF_ADDRESS])
+ if any(
+ Address(existing_override[CONF_ADDRESS]) == address
+ for existing_override in override_config
+ ):
+ raise ValueError("Duplicate override")
+
+ hass.config_entries.async_update_entry(
+ entry=config_entry,
+ options=config_entry.options | {CONF_OVERRIDE: [*override_config, override]},
+ )
+ async_dispatcher_send(hass, SIGNAL_ADD_DEVICE_OVERRIDE, override)
+
+
+def remove_device_override(hass: HomeAssistant, address: Address):
+ """Remove a device override from config."""
+
+ config_entry = get_insteon_config_entry(hass)
+ new_options = {**config_entry.options}
+
+ new_overrides = [
+ existing_override
+ for existing_override in config_entry.options.get(CONF_OVERRIDE, [])
+ if Address(existing_override[CONF_ADDRESS]) != address
+ ]
+ new_options[CONF_OVERRIDE] = new_overrides
+ hass.config_entries.async_update_entry(entry=config_entry, options=new_options)
+
+
+async def _async_connect(**kwargs):
+ """Connect to the Insteon modem."""
+ if devices.modem:
+ await async_close()
+ try:
+ await async_connect(**kwargs)
+ except ConnectionError:
+ return False
+ return True
+
+
+@websocket_api.websocket_command({vol.Required(TYPE): "insteon/config/get"})
+@websocket_api.require_admin
+@websocket_api.async_response
+async def websocket_get_config(
+ hass: HomeAssistant,
+ connection: websocket_api.connection.ActiveConnection,
+ msg: dict[str, Any],
+) -> None:
+ """Get Insteon configuration."""
+ config_entry = get_insteon_config_entry(hass)
+ modem_config = config_entry.data
+ options_config = config_entry.options
+ x10_config = options_config.get(CONF_X10)
+ override_config = options_config.get(CONF_OVERRIDE)
+ connection.send_result(
+ msg[ID],
+ {
+ "modem_config": {**modem_config},
+ "x10_config": x10_config,
+ "override_config": override_config,
+ },
+ )
+
+
+@websocket_api.websocket_command(
+ {
+ vol.Required(TYPE): "insteon/config/get_modem_schema",
+ }
+)
+@websocket_api.require_admin
+@websocket_api.async_response
+async def websocket_get_modem_schema(
+ hass: HomeAssistant,
+ connection: websocket_api.connection.ActiveConnection,
+ msg: dict[str, Any],
+) -> None:
+ """Get the schema for the modem configuration."""
+ config_entry = get_insteon_config_entry(hass)
+ config_data = config_entry.data
+ if device := config_data.get(CONF_DEVICE):
+ ports = await async_get_usb_ports(hass=hass)
+ plm_schema = voluptuous_serialize.convert(
+ build_plm_schema(ports=ports, device=device)
+ )
+ connection.send_result(msg[ID], plm_schema)
+ else:
+ hub_schema = voluptuous_serialize.convert(build_hub_schema(**config_data))
+ connection.send_result(msg[ID], hub_schema)
+
+
+@websocket_api.websocket_command(
+ {
+ vol.Required(TYPE): "insteon/config/update_modem_config",
+ vol.Required("config"): vol.Any(PLM_SCHEMA, HUB_V2_SCHEMA, HUB_V1_SCHEMA),
+ }
+)
+@websocket_api.require_admin
+@websocket_api.async_response
+async def websocket_update_modem_config(
+ hass: HomeAssistant,
+ connection: websocket_api.connection.ActiveConnection,
+ msg: dict[str, Any],
+) -> None:
+ """Get the schema for the modem configuration."""
+ config = msg["config"]
+ config_entry = get_insteon_config_entry(hass)
+ is_connected = devices.modem.connected
+
+ if not await _async_connect(**config):
+ connection.send_error(
+ msg_id=msg[ID], code="connection_failed", message="Connection failed"
+ )
+ # Try to reconnect using old info
+ if is_connected:
+ await _async_connect(**config_entry.data)
+ return
+
+ hass.config_entries.async_update_entry(
+ entry=config_entry,
+ data=config,
+ )
+ connection.send_result(msg[ID], {"status": "success"})
+
+
+@websocket_api.websocket_command(
+ {
+ vol.Required(TYPE): "insteon/config/device_override/add",
+ vol.Required(OVERRIDE): DEVICE_OVERRIDE_SCHEMA,
+ }
+)
+@websocket_api.require_admin
+@websocket_api.async_response
+async def websocket_add_device_override(
+ hass: HomeAssistant,
+ connection: websocket_api.connection.ActiveConnection,
+ msg: dict[str, Any],
+) -> None:
+ """Get the schema for the modem configuration."""
+ override = msg[OVERRIDE]
+ try:
+ add_device_overide(hass, override)
+ except ValueError:
+ connection.send_error(msg[ID], "duplicate", "Duplicate device address")
+
+ connection.send_result(msg[ID])
+
+
+@websocket_api.websocket_command(
+ {
+ vol.Required(TYPE): "insteon/config/device_override/remove",
+ vol.Required(DEVICE_ADDRESS): str,
+ }
+)
+@websocket_api.require_admin
+@websocket_api.async_response
+async def websocket_remove_device_override(
+ hass: HomeAssistant,
+ connection: websocket_api.connection.ActiveConnection,
+ msg: dict[str, Any],
+) -> None:
+ """Get the schema for the modem configuration."""
+ address = Address(msg[DEVICE_ADDRESS])
+ remove_device_override(hass, address)
+ async_dispatcher_send(hass, SIGNAL_REMOVE_DEVICE_OVERRIDE, address)
+ connection.send_result(msg[ID])
diff --git a/homeassistant/components/insteon/api/device.py b/homeassistant/components/insteon/api/device.py
index d48d87fa3475e59e46ff0a7bdbdcd7c4bf2ee5e0..e8bd08bc4eeef7b9052c2c870ff6d7b49e83562b 100644
--- a/homeassistant/components/insteon/api/device.py
+++ b/homeassistant/components/insteon/api/device.py
@@ -3,12 +3,14 @@
from typing import Any
from pyinsteon import devices
+from pyinsteon.address import Address
from pyinsteon.constants import DeviceAction
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.dispatcher import async_dispatcher_send
from ..const import (
DEVICE_ADDRESS,
@@ -18,8 +20,17 @@ from ..const import (
ID,
INSTEON_DEVICE_NOT_FOUND,
MULTIPLE,
+ SIGNAL_REMOVE_HA_DEVICE,
+ SIGNAL_REMOVE_INSTEON_DEVICE,
+ SIGNAL_REMOVE_X10_DEVICE,
TYPE,
)
+from ..schemas import build_x10_schema
+from .config import add_x10_device, remove_device_override, remove_x10_device
+
+X10_DEVICE = "x10_device"
+X10_DEVICE_SCHEMA = build_x10_schema()
+REMOVE_ALL_REFS = "remove_all_refs"
def compute_device_name(ha_device):
@@ -139,3 +150,61 @@ async def websocket_cancel_add_device(
"""Cancel the Insteon all-linking process."""
await devices.async_cancel_all_linking()
connection.send_result(msg[ID])
+
+
+@websocket_api.websocket_command(
+ {
+ vol.Required(TYPE): "insteon/device/remove",
+ vol.Required(DEVICE_ADDRESS): str,
+ vol.Required(REMOVE_ALL_REFS): bool,
+ }
+)
+@websocket_api.require_admin
+@websocket_api.async_response
+async def websocket_remove_device(
+ hass: HomeAssistant,
+ connection: websocket_api.connection.ActiveConnection,
+ msg: dict[str, Any],
+) -> None:
+ """Remove an Insteon device."""
+
+ address = msg[DEVICE_ADDRESS]
+ remove_all_refs = msg[REMOVE_ALL_REFS]
+ if address.startswith("X10"):
+ _, housecode, unitcode = address.split(".")
+ unitcode = int(unitcode)
+ async_dispatcher_send(hass, SIGNAL_REMOVE_X10_DEVICE, housecode, unitcode)
+ remove_x10_device(hass, housecode, unitcode)
+ else:
+ address = Address(address)
+ remove_device_override(hass, address)
+ async_dispatcher_send(hass, SIGNAL_REMOVE_HA_DEVICE, address)
+ async_dispatcher_send(
+ hass, SIGNAL_REMOVE_INSTEON_DEVICE, address, remove_all_refs
+ )
+
+ connection.send_result(msg[ID])
+
+
+@websocket_api.websocket_command(
+ {
+ vol.Required(TYPE): "insteon/device/add_x10",
+ vol.Required(X10_DEVICE): X10_DEVICE_SCHEMA,
+ }
+)
+@websocket_api.require_admin
+@websocket_api.async_response
+async def websocket_add_x10_device(
+ hass: HomeAssistant,
+ connection: websocket_api.connection.ActiveConnection,
+ msg: dict[str, Any],
+) -> None:
+ """Get the schema for the X10 devices configuration."""
+ x10_device = msg[X10_DEVICE]
+ try:
+ add_x10_device(hass, x10_device)
+ except ValueError:
+ connection.send_error(msg[ID], code="duplicate", message="Duplicate X10 device")
+ return
+
+ connection.send_result(msg[ID])
diff --git a/homeassistant/components/insteon/config_flow.py b/homeassistant/components/insteon/config_flow.py
index 44aa1e18646fb10c162a0e977140c8f00d21c33f..baf06b138608a0a4cec6a2bf5f8b6c4296c59c05 100644
--- a/homeassistant/components/insteon/config_flow.py
+++ b/homeassistant/components/insteon/config_flow.py
@@ -4,52 +4,19 @@ from __future__ import annotations
import logging
-from pyinsteon import async_close, async_connect, devices
+from pyinsteon import async_connect
from homeassistant.components import dhcp, usb
from homeassistant.config_entries import (
DEFAULT_DISCOVERY_UNIQUE_ID,
- ConfigEntry,
ConfigFlow,
ConfigFlowResult,
- OptionsFlow,
)
-from homeassistant.const import (
- CONF_ADDRESS,
- CONF_DEVICE,
- CONF_HOST,
- CONF_NAME,
- CONF_PASSWORD,
- CONF_PORT,
- CONF_USERNAME,
-)
-from homeassistant.core import callback
+from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_NAME
from homeassistant.helpers.device_registry import format_mac
-from homeassistant.helpers.dispatcher import async_dispatcher_send
-from .const import (
- CONF_HOUSECODE,
- CONF_HUB_VERSION,
- CONF_OVERRIDE,
- CONF_UNITCODE,
- CONF_X10,
- DOMAIN,
- SIGNAL_ADD_DEVICE_OVERRIDE,
- SIGNAL_ADD_X10_DEVICE,
- SIGNAL_REMOVE_DEVICE_OVERRIDE,
- SIGNAL_REMOVE_X10_DEVICE,
-)
-from .schemas import (
- add_device_override,
- add_x10_device,
- build_device_override_schema,
- build_hub_schema,
- build_plm_manual_schema,
- build_plm_schema,
- build_remove_override_schema,
- build_remove_x10_schema,
- build_x10_schema,
-)
+from .const import CONF_HUB_VERSION, DOMAIN
+from .schemas import build_hub_schema, build_plm_manual_schema, build_plm_schema
from .utils import async_get_usb_ports
STEP_PLM = "plm"
@@ -80,41 +47,6 @@ async def _async_connect(**kwargs):
return True
-def _remove_override(address, options):
- """Remove a device override from config."""
- new_options = {}
- if options.get(CONF_X10):
- new_options[CONF_X10] = options.get(CONF_X10)
- new_overrides = [
- override
- for override in options[CONF_OVERRIDE]
- if override[CONF_ADDRESS] != address
- ]
- if new_overrides:
- new_options[CONF_OVERRIDE] = new_overrides
- return new_options
-
-
-def _remove_x10(device, options):
- """Remove an X10 device from the config."""
- housecode = device[11].lower()
- unitcode = int(device[24:])
- new_options = {}
- if options.get(CONF_OVERRIDE):
- new_options[CONF_OVERRIDE] = options.get(CONF_OVERRIDE)
- new_x10 = [
- existing_device
- for existing_device in options[CONF_X10]
- if (
- existing_device[CONF_HOUSECODE].lower() != housecode
- or existing_device[CONF_UNITCODE] != unitcode
- )
- ]
- if new_x10:
- new_options[CONF_X10] = new_x10
- return new_options, housecode, unitcode
-
-
class InsteonFlowHandler(ConfigFlow, domain=DOMAIN):
"""Insteon config flow handler."""
@@ -122,14 +54,6 @@ class InsteonFlowHandler(ConfigFlow, domain=DOMAIN):
_device_name: str | None = None
discovered_conf: dict[str, str] = {}
- @staticmethod
- @callback
- def async_get_options_flow(
- config_entry: ConfigEntry,
- ) -> InsteonOptionsFlowHandler:
- """Define the config flow to handle options."""
- return InsteonOptionsFlowHandler(config_entry)
-
async def async_step_user(self, user_input=None):
"""Init the config flow."""
if self._async_current_entries():
@@ -237,140 +161,3 @@ class InsteonFlowHandler(ConfigFlow, domain=DOMAIN):
}
await self.async_set_unique_id(format_mac(discovery_info.macaddress))
return await self.async_step_user()
-
-
-class InsteonOptionsFlowHandler(OptionsFlow):
- """Handle an Insteon options flow."""
-
- def __init__(self, config_entry: ConfigEntry) -> None:
- """Init the InsteonOptionsFlowHandler class."""
- self.config_entry = config_entry
-
- async def async_step_init(self, user_input=None) -> ConfigFlowResult:
- """Init the options config flow."""
- menu_options = [STEP_ADD_OVERRIDE, STEP_ADD_X10]
-
- if self.config_entry.data.get(CONF_HOST):
- menu_options.append(STEP_CHANGE_HUB_CONFIG)
- else:
- menu_options.append(STEP_CHANGE_PLM_CONFIG)
-
- options = {**self.config_entry.options}
- if options.get(CONF_OVERRIDE):
- menu_options.append(STEP_REMOVE_OVERRIDE)
- if options.get(CONF_X10):
- menu_options.append(STEP_REMOVE_X10)
-
- return self.async_show_menu(step_id="init", menu_options=menu_options)
-
- async def async_step_change_hub_config(self, user_input=None) -> ConfigFlowResult:
- """Change the Hub configuration."""
- errors = {}
- if user_input is not None:
- data = {
- **self.config_entry.data,
- CONF_HOST: user_input[CONF_HOST],
- CONF_PORT: user_input[CONF_PORT],
- }
- if self.config_entry.data[CONF_HUB_VERSION] == 2:
- data[CONF_USERNAME] = user_input[CONF_USERNAME]
- data[CONF_PASSWORD] = user_input[CONF_PASSWORD]
- 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, errors=errors
- )
-
- async def async_step_change_plm_config(self, user_input=None) -> ConfigFlowResult:
- """Change the PLM configuration."""
- errors = {}
- if user_input is not None:
- data = {
- **self.config_entry.data,
- CONF_DEVICE: user_input[CONF_DEVICE],
- }
- 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, errors=errors
- )
-
- async def async_step_add_override(self, user_input=None) -> ConfigFlowResult:
- """Add a device override."""
- errors = {}
- if user_input is not None:
- 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(data=data)
- except ValueError:
- errors["base"] = "input_error"
- schema_defaults = user_input if user_input is not None else {}
- data_schema = build_device_override_schema(**schema_defaults)
- return self.async_show_form(
- step_id=STEP_ADD_OVERRIDE, data_schema=data_schema, errors=errors
- )
-
- async def async_step_add_x10(self, user_input=None) -> ConfigFlowResult:
- """Add an X10 device."""
- errors: dict[str, str] = {}
- 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(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(
- step_id=STEP_ADD_X10, data_schema=data_schema, errors=errors
- )
-
- async def async_step_remove_override(self, user_input=None) -> ConfigFlowResult:
- """Remove a device override."""
- errors: dict[str, str] = {}
- options = self.config_entry.options
- if user_input is not None:
- options = _remove_override(user_input[CONF_ADDRESS], options)
- async_dispatcher_send(
- self.hass,
- SIGNAL_REMOVE_DEVICE_OVERRIDE,
- user_input[CONF_ADDRESS],
- )
- return self.async_create_entry(data=options)
-
- data_schema = build_remove_override_schema(options[CONF_OVERRIDE])
- return self.async_show_form(
- step_id=STEP_REMOVE_OVERRIDE, data_schema=data_schema, errors=errors
- )
-
- async def async_step_remove_x10(self, user_input=None) -> ConfigFlowResult:
- """Remove an X10 device."""
- errors: dict[str, str] = {}
- options = self.config_entry.options
- if user_input is not None:
- options, housecode, unitcode = _remove_x10(user_input[CONF_DEVICE], options)
- async_dispatcher_send(
- self.hass, SIGNAL_REMOVE_X10_DEVICE, housecode, unitcode
- )
- return self.async_create_entry(data=options)
-
- data_schema = build_remove_x10_schema(options[CONF_X10])
- return self.async_show_form(
- step_id=STEP_REMOVE_X10, data_schema=data_schema, errors=errors
- )
diff --git a/homeassistant/components/insteon/const.py b/homeassistant/components/insteon/const.py
index b7e6e6055e1593270d83a475a70316dd5910c513..11e1943aa736d6eab3c938b50dc226eccd3f5d29 100644
--- a/homeassistant/components/insteon/const.py
+++ b/homeassistant/components/insteon/const.py
@@ -101,6 +101,8 @@ SIGNAL_SAVE_DEVICES = "save_devices"
SIGNAL_ADD_ENTITIES = "insteon_add_entities"
SIGNAL_ADD_DEFAULT_LINKS = "add_default_links"
SIGNAL_ADD_DEVICE_OVERRIDE = "add_device_override"
+SIGNAL_REMOVE_HA_DEVICE = "insteon_remove_ha_device"
+SIGNAL_REMOVE_INSTEON_DEVICE = "insteon_remove_insteon_device"
SIGNAL_REMOVE_DEVICE_OVERRIDE = "insteon_remove_device_override"
SIGNAL_REMOVE_ENTITY = "insteon_remove_entity"
SIGNAL_ADD_X10_DEVICE = "insteon_add_x10_device"
diff --git a/homeassistant/components/insteon/insteon_entity.py b/homeassistant/components/insteon/insteon_entity.py
index f81298dfe481184227ab03911eee4d586b104d18..79e5c18a934de48fac96bc915ad9568263adfcb0 100644
--- a/homeassistant/components/insteon/insteon_entity.py
+++ b/homeassistant/components/insteon/insteon_entity.py
@@ -95,6 +95,7 @@ class InsteonEntity(Entity):
f" {self._insteon_device.engine_version}"
),
via_device=(DOMAIN, str(devices.modem.address)),
+ configuration_url=f"homeassistant://insteon/device/config/{self._insteon_device.id}",
)
@callback
diff --git a/homeassistant/components/insteon/manifest.json b/homeassistant/components/insteon/manifest.json
index cf2109638416fa445bffbd15ee62294c988b542c..7d12436d0fb5ac2bf2e0e1eb58af90e098f646e5 100644
--- a/homeassistant/components/insteon/manifest.json
+++ b/homeassistant/components/insteon/manifest.json
@@ -18,7 +18,7 @@
"loggers": ["pyinsteon", "pypubsub"],
"requirements": [
"pyinsteon==1.5.3",
- "insteon-frontend-home-assistant==0.4.0"
+ "insteon-frontend-home-assistant==0.5.0"
],
"usb": [
{
diff --git a/homeassistant/components/insteon/schemas.py b/homeassistant/components/insteon/schemas.py
index e277281c240c59ff25792434818723379d42bd45..837c6224014e54f76d3e2f57ba1714ef7d9b2dfb 100644
--- a/homeassistant/components/insteon/schemas.py
+++ b/homeassistant/components/insteon/schemas.py
@@ -2,9 +2,6 @@
from __future__ import annotations
-from binascii import Error as HexError, unhexlify
-
-from pyinsteon.address import Address
from pyinsteon.constants import HC_LOOKUP
import voluptuous as vol
@@ -25,10 +22,8 @@ from .const import (
CONF_CAT,
CONF_DIM_STEPS,
CONF_HOUSECODE,
- CONF_OVERRIDE,
CONF_SUBCAT,
CONF_UNITCODE,
- CONF_X10,
HOUSECODES,
PORT_HUB_V1,
PORT_HUB_V2,
@@ -76,76 +71,6 @@ TRIGGER_SCENE_SCHEMA = vol.Schema(
ADD_DEFAULT_LINKS_SCHEMA = vol.Schema({vol.Required(CONF_ENTITY_ID): cv.entity_id})
-def normalize_byte_entry_to_int(entry: int | bytes | str):
- """Format a hex entry value."""
- if isinstance(entry, int):
- if entry in range(256):
- return entry
- raise ValueError("Must be single byte")
- if isinstance(entry, str):
- if entry[0:2].lower() == "0x":
- entry = entry[2:]
- if len(entry) != 2:
- raise ValueError("Not a valid hex code")
- try:
- entry = unhexlify(entry)
- except HexError as err:
- raise ValueError("Not a valid hex code") from err
- return int.from_bytes(entry, byteorder="big")
-
-
-def add_device_override(config_data, new_override):
- """Add a new device override."""
- try:
- address = str(Address(new_override[CONF_ADDRESS]))
- cat = normalize_byte_entry_to_int(new_override[CONF_CAT])
- subcat = normalize_byte_entry_to_int(new_override[CONF_SUBCAT])
- except ValueError as err:
- raise ValueError("Incorrect values") from err
-
- overrides = [
- override
- for override in config_data.get(CONF_OVERRIDE, [])
- if override[CONF_ADDRESS] != address
- ]
- overrides.append(
- {
- CONF_ADDRESS: address,
- CONF_CAT: cat,
- CONF_SUBCAT: subcat,
- }
- )
-
- new_config = {}
- if config_data.get(CONF_X10):
- new_config[CONF_X10] = config_data[CONF_X10]
- new_config[CONF_OVERRIDE] = overrides
- return new_config
-
-
-def add_x10_device(config_data, new_x10):
- """Add a new X10 device to X10 device list."""
- x10_devices = [
- x10_device
- for x10_device in config_data.get(CONF_X10, [])
- if x10_device[CONF_HOUSECODE] != new_x10[CONF_HOUSECODE]
- or x10_device[CONF_UNITCODE] != new_x10[CONF_UNITCODE]
- ]
- x10_devices.append(
- {
- CONF_HOUSECODE: new_x10[CONF_HOUSECODE],
- CONF_UNITCODE: new_x10[CONF_UNITCODE],
- CONF_PLATFORM: new_x10[CONF_PLATFORM],
- CONF_DIM_STEPS: new_x10[CONF_DIM_STEPS],
- }
- )
- new_config = {}
- if config_data.get(CONF_OVERRIDE):
- new_config[CONF_OVERRIDE] = config_data[CONF_OVERRIDE]
- new_config[CONF_X10] = x10_devices
- return new_config
-
-
def build_device_override_schema(
address=vol.UNDEFINED,
cat=vol.UNDEFINED,
@@ -169,12 +94,16 @@ def build_x10_schema(
dim_steps=22,
):
"""Build the X10 schema for config flow."""
+ if platform == "light":
+ dim_steps_schema = vol.Required(CONF_DIM_STEPS, default=dim_steps)
+ else:
+ dim_steps_schema = vol.Optional(CONF_DIM_STEPS, default=dim_steps)
return vol.Schema(
{
vol.Required(CONF_HOUSECODE, default=housecode): vol.In(HC_LOOKUP.keys()),
vol.Required(CONF_UNITCODE, default=unitcode): vol.In(range(1, 17)),
vol.Required(CONF_PLATFORM, default=platform): vol.In(X10_PLATFORMS),
- vol.Optional(CONF_DIM_STEPS, default=dim_steps): vol.In(range(1, 255)),
+ dim_steps_schema: vol.Range(min=0, max=255),
}
)
@@ -219,18 +148,3 @@ def build_hub_schema(
schema[vol.Required(CONF_USERNAME, default=username)] = str
schema[vol.Required(CONF_PASSWORD, default=password)] = str
return vol.Schema(schema)
-
-
-def build_remove_override_schema(data):
- """Build the schema to remove device overrides in config flow options."""
- selection = [override[CONF_ADDRESS] for override in data]
- return vol.Schema({vol.Required(CONF_ADDRESS): vol.In(selection)})
-
-
-def build_remove_x10_schema(data):
- """Build the schema to remove an X10 device in config flow options."""
- selection = [
- f"Housecode: {device[CONF_HOUSECODE].upper()}, Unitcode: {device[CONF_UNITCODE]}"
- for device in data
- ]
- return vol.Schema({vol.Required(CONF_DEVICE): vol.In(selection)})
diff --git a/homeassistant/components/insteon/utils.py b/homeassistant/components/insteon/utils.py
index 272018ea5072b932ce1e090c4a8b5d4ca310a703..db25d8c97a9b7e39382b103fd5091113fc353fde 100644
--- a/homeassistant/components/insteon/utils.py
+++ b/homeassistant/components/insteon/utils.py
@@ -65,6 +65,8 @@ from .const import (
SIGNAL_PRINT_ALDB,
SIGNAL_REMOVE_DEVICE_OVERRIDE,
SIGNAL_REMOVE_ENTITY,
+ SIGNAL_REMOVE_HA_DEVICE,
+ SIGNAL_REMOVE_INSTEON_DEVICE,
SIGNAL_REMOVE_X10_DEVICE,
SIGNAL_SAVE_DEVICES,
SRV_ADD_ALL_LINK,
@@ -179,7 +181,7 @@ def register_new_device_callback(hass):
@callback
-def async_register_services(hass):
+def async_register_services(hass): # noqa: C901
"""Register services used by insteon component."""
save_lock = asyncio.Lock()
@@ -270,14 +272,14 @@ def async_register_services(hass):
async def async_add_device_override(override):
"""Remove an Insten device and associated entities."""
address = Address(override[CONF_ADDRESS])
- await async_remove_device(address)
+ await async_remove_ha_device(address)
devices.set_id(address, override[CONF_CAT], override[CONF_SUBCAT], 0)
await async_srv_save_devices()
async def async_remove_device_override(address):
"""Remove an Insten device and associated entities."""
address = Address(address)
- await async_remove_device(address)
+ await async_remove_ha_device(address)
devices.set_id(address, None, None, None)
await devices.async_identify_device(address)
await async_srv_save_devices()
@@ -304,9 +306,9 @@ def async_register_services(hass):
"""Remove an X10 device and associated entities."""
address = create_x10_address(housecode, unitcode)
devices.pop(address)
- await async_remove_device(address)
+ await async_remove_ha_device(address)
- async def async_remove_device(address):
+ async def async_remove_ha_device(address: Address, remove_all_refs: bool = False):
"""Remove the device and all entities from hass."""
signal = f"{address.id}_{SIGNAL_REMOVE_ENTITY}"
async_dispatcher_send(hass, signal)
@@ -315,6 +317,15 @@ def async_register_services(hass):
if device:
dev_registry.async_remove_device(device.id)
+ async def async_remove_insteon_device(
+ address: Address, remove_all_refs: bool = False
+ ):
+ """Remove the underlying Insteon device from the network."""
+ await devices.async_remove_device(
+ address=address, force=False, remove_all_refs=remove_all_refs
+ )
+ await async_srv_save_devices()
+
hass.services.async_register(
DOMAIN, SRV_ADD_ALL_LINK, async_srv_add_all_link, schema=ADD_ALL_LINK_SCHEMA
)
@@ -368,6 +379,10 @@ def async_register_services(hass):
)
async_dispatcher_connect(hass, SIGNAL_ADD_X10_DEVICE, async_add_x10_device)
async_dispatcher_connect(hass, SIGNAL_REMOVE_X10_DEVICE, async_remove_x10_device)
+ async_dispatcher_connect(hass, SIGNAL_REMOVE_HA_DEVICE, async_remove_ha_device)
+ async_dispatcher_connect(
+ hass, SIGNAL_REMOVE_INSTEON_DEVICE, async_remove_insteon_device
+ )
_LOGGER.debug("Insteon Services registered")
diff --git a/requirements_all.txt b/requirements_all.txt
index 7a9a99c26286dc22df4ffce4c4199dbd41b53873..653e481d2fb08ff98b3e4a2f18abfac78a4c88b3 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -1142,7 +1142,7 @@ influxdb==5.3.1
inkbird-ble==0.5.6
# homeassistant.components.insteon
-insteon-frontend-home-assistant==0.4.0
+insteon-frontend-home-assistant==0.5.0
# homeassistant.components.intellifire
intellifire4py==2.2.2
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index 6172bb9700717709d7202e41113b6ef267d24f48..0decf82fe0cb7e7c6388a31590aca8c70ea8063d 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -926,7 +926,7 @@ influxdb==5.3.1
inkbird-ble==0.5.6
# homeassistant.components.insteon
-insteon-frontend-home-assistant==0.4.0
+insteon-frontend-home-assistant==0.5.0
# homeassistant.components.intellifire
intellifire4py==2.2.2
diff --git a/tests/components/insteon/mock_setup.py b/tests/components/insteon/mock_setup.py
new file mode 100644
index 0000000000000000000000000000000000000000..c0d90509a50dc6c97b856fd43cac8eff0d8ce513
--- /dev/null
+++ b/tests/components/insteon/mock_setup.py
@@ -0,0 +1,44 @@
+"""Utility to setup the Insteon integration."""
+
+from homeassistant.components.insteon.api import async_load_api
+from homeassistant.components.insteon.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+
+from .const import MOCK_USER_INPUT_PLM
+from .mock_devices import MockDevices
+
+from tests.common import MockConfigEntry
+from tests.typing import WebSocketGenerator
+
+
+async def async_mock_setup(
+ hass: HomeAssistant,
+ hass_ws_client: WebSocketGenerator,
+ config_data: dict | None = None,
+ config_options: dict | None = None,
+):
+ """Set up for tests."""
+ config_data = MOCK_USER_INPUT_PLM if config_data is None else config_data
+ config_options = {} if config_options is None else config_options
+ config_entry = MockConfigEntry(
+ domain=DOMAIN,
+ entry_id="abcde12345",
+ data=config_data,
+ options=config_options,
+ )
+ config_entry.add_to_hass(hass)
+ async_load_api(hass)
+
+ ws_client = await hass_ws_client(hass)
+ devices = MockDevices()
+ await devices.async_load()
+
+ dev_reg = dr.async_get(hass)
+ # Create device registry entry for mock node
+ ha_device = dev_reg.async_get_or_create(
+ config_entry_id=config_entry.entry_id,
+ identifiers={(DOMAIN, "11.11.11")},
+ name="Device 11.11.11",
+ )
+ return ws_client, devices, ha_device, dev_reg
diff --git a/tests/components/insteon/test_api_config.py b/tests/components/insteon/test_api_config.py
new file mode 100644
index 0000000000000000000000000000000000000000..7c922338638f5aa791e0275fd70a28e14d888eef
--- /dev/null
+++ b/tests/components/insteon/test_api_config.py
@@ -0,0 +1,391 @@
+"""Test the Insteon APIs for configuring the integration."""
+
+from unittest.mock import patch
+
+from homeassistant.components.insteon.api.device import ID, TYPE
+from homeassistant.components.insteon.const import (
+ CONF_HUB_VERSION,
+ CONF_OVERRIDE,
+ CONF_X10,
+)
+from homeassistant.core import HomeAssistant
+
+from .const import (
+ MOCK_DEVICE,
+ MOCK_HOSTNAME,
+ MOCK_USER_INPUT_HUB_V1,
+ MOCK_USER_INPUT_HUB_V2,
+ MOCK_USER_INPUT_PLM,
+)
+from .mock_connection import mock_failed_connection, mock_successful_connection
+from .mock_setup import async_mock_setup
+
+from tests.typing import WebSocketGenerator
+
+
+class MockProtocol:
+ """A mock Insteon protocol object."""
+
+ connected = True
+
+
+async def test_get_config(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test getting the Insteon configuration."""
+
+ ws_client, _, _, _ = await async_mock_setup(hass, hass_ws_client)
+ await ws_client.send_json({ID: 2, TYPE: "insteon/config/get"})
+ msg = await ws_client.receive_json()
+ result = msg["result"]
+
+ assert result["modem_config"] == {"device": MOCK_DEVICE}
+
+
+async def test_get_modem_schema_plm(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test getting the Insteon PLM modem configuration schema."""
+
+ ws_client, _, _, _ = await async_mock_setup(hass, hass_ws_client)
+ await ws_client.send_json({ID: 2, TYPE: "insteon/config/get_modem_schema"})
+ msg = await ws_client.receive_json()
+ result = msg["result"][0]
+
+ assert result["default"] == MOCK_DEVICE
+ assert result["name"] == "device"
+ assert result["required"]
+
+
+async def test_get_modem_schema_hub(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test getting the Insteon PLM modem configuration schema."""
+
+ ws_client, devices, _, _ = await async_mock_setup(
+ hass,
+ hass_ws_client,
+ config_data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
+ )
+ await ws_client.send_json({ID: 2, TYPE: "insteon/config/get_modem_schema"})
+ msg = await ws_client.receive_json()
+ result = msg["result"][0]
+
+ assert result["default"] == MOCK_HOSTNAME
+ assert result["name"] == "host"
+ assert result["required"]
+
+
+async def test_update_modem_config_plm(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test getting the Insteon PLM modem configuration schema."""
+
+ ws_client, mock_devices, _, _ = await async_mock_setup(
+ hass,
+ hass_ws_client,
+ )
+ with (
+ patch(
+ "homeassistant.components.insteon.api.config.async_connect",
+ new=mock_successful_connection,
+ ),
+ patch("homeassistant.components.insteon.api.config.devices", mock_devices),
+ patch("homeassistant.components.insteon.api.config.async_close"),
+ ):
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/config/update_modem_config",
+ "config": MOCK_USER_INPUT_PLM,
+ }
+ )
+ msg = await ws_client.receive_json()
+ result = msg["result"]
+
+ assert result["status"] == "success"
+
+
+async def test_update_modem_config_hub_v2(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test getting the Insteon HubV2 modem configuration schema."""
+
+ ws_client, mock_devices, _, _ = await async_mock_setup(
+ hass,
+ hass_ws_client,
+ config_data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
+ config_options={"dev_path": "/some/path"},
+ )
+ with (
+ patch(
+ "homeassistant.components.insteon.api.config.async_connect",
+ new=mock_successful_connection,
+ ),
+ patch("homeassistant.components.insteon.api.config.devices", mock_devices),
+ patch("homeassistant.components.insteon.api.config.async_close"),
+ ):
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/config/update_modem_config",
+ "config": MOCK_USER_INPUT_HUB_V2,
+ }
+ )
+ msg = await ws_client.receive_json()
+ result = msg["result"]
+
+ assert result["status"] == "success"
+
+
+async def test_update_modem_config_hub_v1(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test getting the Insteon HubV1 modem configuration schema."""
+
+ ws_client, mock_devices, _, _ = await async_mock_setup(
+ hass,
+ hass_ws_client,
+ config_data={**MOCK_USER_INPUT_HUB_V1, CONF_HUB_VERSION: 1},
+ )
+ with (
+ patch(
+ "homeassistant.components.insteon.api.config.async_connect",
+ new=mock_successful_connection,
+ ),
+ patch("homeassistant.components.insteon.api.config.devices", mock_devices),
+ patch("homeassistant.components.insteon.api.config.async_close"),
+ ):
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/config/update_modem_config",
+ "config": MOCK_USER_INPUT_HUB_V1,
+ }
+ )
+ msg = await ws_client.receive_json()
+ result = msg["result"]
+
+ assert result["status"] == "success"
+
+
+async def test_update_modem_config_bad(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test updating the Insteon modem configuration with bad connection information."""
+
+ ws_client, mock_devices, _, _ = await async_mock_setup(
+ hass,
+ hass_ws_client,
+ )
+ with (
+ patch(
+ "homeassistant.components.insteon.api.config.async_connect",
+ new=mock_failed_connection,
+ ),
+ patch("homeassistant.components.insteon.api.config.devices", mock_devices),
+ patch("homeassistant.components.insteon.api.config.async_close"),
+ ):
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/config/update_modem_config",
+ "config": MOCK_USER_INPUT_PLM,
+ }
+ )
+ msg = await ws_client.receive_json()
+ result = msg["error"]
+ assert result["code"] == "connection_failed"
+
+
+async def test_update_modem_config_bad_reconnect(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test updating the Insteon modem configuration with bad connection information so reconnect to old."""
+
+ ws_client, mock_devices, _, _ = await async_mock_setup(
+ hass,
+ hass_ws_client,
+ )
+ with (
+ patch(
+ "homeassistant.components.insteon.api.config.async_connect",
+ new=mock_failed_connection,
+ ),
+ patch("homeassistant.components.insteon.api.config.devices", mock_devices),
+ patch("homeassistant.components.insteon.api.config.async_close"),
+ ):
+ mock_devices.modem.protocol = MockProtocol()
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/config/update_modem_config",
+ "config": MOCK_USER_INPUT_PLM,
+ }
+ )
+ msg = await ws_client.receive_json()
+ result = msg["error"]
+ assert result["code"] == "connection_failed"
+
+
+async def test_add_device_override(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test adding a device configuration override."""
+
+ ws_client, _, _, _ = await async_mock_setup(hass, hass_ws_client)
+ override = {
+ "address": "99.99.99",
+ "cat": "0x01",
+ "subcat": "0x03",
+ }
+ await ws_client.send_json(
+ {ID: 2, TYPE: "insteon/config/device_override/add", "override": override}
+ )
+ msg = await ws_client.receive_json()
+ assert msg["success"]
+
+ config_entry = hass.config_entries.async_get_entry("abcde12345")
+ assert len(config_entry.options[CONF_OVERRIDE]) == 1
+ assert config_entry.options[CONF_OVERRIDE][0]["address"] == "99.99.99"
+
+
+async def test_add_device_override_duplicate(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test adding a duplicate device configuration override."""
+
+ override = {
+ "address": "99.99.99",
+ "cat": "0x01",
+ "subcat": "0x03",
+ }
+
+ ws_client, _, _, _ = await async_mock_setup(
+ hass, hass_ws_client, config_options={CONF_OVERRIDE: [override]}
+ )
+ await ws_client.send_json(
+ {ID: 2, TYPE: "insteon/config/device_override/add", "override": override}
+ )
+ msg = await ws_client.receive_json()
+ assert msg["error"]
+
+
+async def test_remove_device_override(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test removing a device configuration override."""
+
+ override = {
+ "address": "99.99.99",
+ "cat": "0x01",
+ "subcat": "0x03",
+ }
+ overrides = [
+ override,
+ {
+ "address": "88.88.88",
+ "cat": "0x02",
+ "subcat": "0x05",
+ },
+ ]
+
+ ws_client, _, _, _ = await async_mock_setup(
+ hass, hass_ws_client, config_options={CONF_OVERRIDE: overrides}
+ )
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/config/device_override/remove",
+ "device_address": "99.99.99",
+ }
+ )
+ msg = await ws_client.receive_json()
+ assert msg["success"]
+
+ config_entry = hass.config_entries.async_get_entry("abcde12345")
+ assert len(config_entry.options[CONF_OVERRIDE]) == 1
+ assert config_entry.options[CONF_OVERRIDE][0]["address"] == "88.88.88"
+
+
+async def test_add_device_override_with_x10(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test adding a device configuration override when X10 configuration exists."""
+
+ x10_device = {"housecode": "a", "unitcode": 1, "platform": "switch"}
+ ws_client, _, _, _ = await async_mock_setup(
+ hass, hass_ws_client, config_options={CONF_X10: [x10_device]}
+ )
+ override = {
+ "address": "99.99.99",
+ "cat": "0x01",
+ "subcat": "0x03",
+ }
+ await ws_client.send_json(
+ {ID: 2, TYPE: "insteon/config/device_override/add", "override": override}
+ )
+ msg = await ws_client.receive_json()
+ assert msg["success"]
+
+ config_entry = hass.config_entries.async_get_entry("abcde12345")
+ assert len(config_entry.options[CONF_X10]) == 1
+
+
+async def test_remove_device_override_with_x10(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test removing a device configuration override when X10 configuration exists."""
+
+ override = {
+ "address": "99.99.99",
+ "cat": "0x01",
+ "subcat": "0x03",
+ }
+ overrides = [
+ override,
+ {
+ "address": "88.88.88",
+ "cat": "0x02",
+ "subcat": "0x05",
+ },
+ ]
+ x10_device = {"housecode": "a", "unitcode": 1, "platform": "switch"}
+
+ ws_client, _, _, _ = await async_mock_setup(
+ hass,
+ hass_ws_client,
+ config_options={CONF_OVERRIDE: overrides, CONF_X10: [x10_device]},
+ )
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/config/device_override/remove",
+ "device_address": "99.99.99",
+ }
+ )
+ msg = await ws_client.receive_json()
+ assert msg["success"]
+
+ config_entry = hass.config_entries.async_get_entry("abcde12345")
+ assert len(config_entry.options[CONF_X10]) == 1
+
+
+async def test_remove_device_override_no_overrides(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test removing a device override when no overrides are configured."""
+
+ ws_client, _, _, _ = await async_mock_setup(hass, hass_ws_client)
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/config/device_override/remove",
+ "device_address": "99.99.99",
+ }
+ )
+ msg = await ws_client.receive_json()
+ assert msg["success"]
+
+ config_entry = hass.config_entries.async_get_entry("abcde12345")
+ assert not config_entry.options.get(CONF_OVERRIDE)
diff --git a/tests/components/insteon/test_api_device.py b/tests/components/insteon/test_api_device.py
index f3c67d479d0711095d4e1b41509d18fc2cdeddc4..29d601eb3effd42516285fcb5bcbee3366f322b7 100644
--- a/tests/components/insteon/test_api_device.py
+++ b/tests/components/insteon/test_api_device.py
@@ -18,48 +18,29 @@ from homeassistant.components.insteon.api.device import (
TYPE,
async_device_name,
)
-from homeassistant.components.insteon.const import DOMAIN, MULTIPLE
+from homeassistant.components.insteon.const import (
+ CONF_OVERRIDE,
+ CONF_X10,
+ DOMAIN,
+ MULTIPLE,
+)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from .const import MOCK_USER_INPUT_PLM
from .mock_devices import MockDevices
+from .mock_setup import async_mock_setup
from tests.common import MockConfigEntry
from tests.typing import WebSocketGenerator
-async def _async_setup(hass, hass_ws_client):
- """Set up for tests."""
- config_entry = MockConfigEntry(
- domain=DOMAIN,
- entry_id="abcde12345",
- data=MOCK_USER_INPUT_PLM,
- options={},
- )
- config_entry.add_to_hass(hass)
- async_load_api(hass)
-
- ws_client = await hass_ws_client(hass)
- devices = MockDevices()
- await devices.async_load()
-
- dev_reg = dr.async_get(hass)
- # Create device registry entry for mock node
- ha_device = dev_reg.async_get_or_create(
- config_entry_id=config_entry.entry_id,
- identifiers={(DOMAIN, "11.11.11")},
- name="Device 11.11.11",
- )
- return ws_client, devices, ha_device, dev_reg
-
-
-async def test_get_device_api(
+async def test_get_config(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test getting an Insteon device."""
- ws_client, devices, ha_device, _ = await _async_setup(hass, hass_ws_client)
+ ws_client, devices, ha_device, _ = await async_mock_setup(hass, hass_ws_client)
with patch.object(insteon.api.device, "devices", devices):
await ws_client.send_json(
{ID: 2, TYPE: "insteon/device/get", DEVICE_ID: ha_device.id}
@@ -76,7 +57,7 @@ async def test_no_ha_device(
) -> None:
"""Test response when no HA device exists."""
- ws_client, devices, _, _ = await _async_setup(hass, hass_ws_client)
+ ws_client, devices, _, _ = await async_mock_setup(hass, hass_ws_client)
with patch.object(insteon.api.device, "devices", devices):
await ws_client.send_json(
{ID: 2, TYPE: "insteon/device/get", DEVICE_ID: "not_a_device"}
@@ -141,7 +122,7 @@ async def test_get_ha_device_name(
) -> None:
"""Test getting the HA device name from an Insteon address."""
- _, devices, _, device_reg = await _async_setup(hass, hass_ws_client)
+ _, devices, _, device_reg = await async_mock_setup(hass, hass_ws_client)
with patch.object(insteon.api.device, "devices", devices):
# Test a real HA and Insteon device
@@ -164,7 +145,7 @@ async def test_add_device_api(
) -> None:
"""Test adding an Insteon device."""
- ws_client, devices, _, _ = await _async_setup(hass, hass_ws_client)
+ ws_client, devices, _, _ = await async_mock_setup(hass, hass_ws_client)
with patch.object(insteon.api.device, "devices", devices):
await ws_client.send_json({ID: 2, TYPE: "insteon/device/add", MULTIPLE: True})
@@ -194,7 +175,7 @@ async def test_cancel_add_device(
) -> None:
"""Test cancelling adding of a new device."""
- ws_client, devices, _, _ = await _async_setup(hass, hass_ws_client)
+ ws_client, devices, _, _ = await async_mock_setup(hass, hass_ws_client)
with patch.object(insteon.api.aldb, "devices", devices):
await ws_client.send_json(
@@ -205,3 +186,127 @@ async def test_cancel_add_device(
)
msg = await ws_client.receive_json()
assert msg["success"]
+
+
+async def test_add_x10_device(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test adding an X10 device."""
+
+ ws_client, _, _, _ = await async_mock_setup(hass, hass_ws_client)
+ x10_device = {"housecode": "a", "unitcode": 1, "platform": "switch"}
+ await ws_client.send_json(
+ {ID: 2, TYPE: "insteon/device/add_x10", "x10_device": x10_device}
+ )
+ msg = await ws_client.receive_json()
+ assert msg["success"]
+
+ config_entry = hass.config_entries.async_get_entry("abcde12345")
+ assert len(config_entry.options[CONF_X10]) == 1
+ assert config_entry.options[CONF_X10][0]["housecode"] == "a"
+ assert config_entry.options[CONF_X10][0]["unitcode"] == 1
+ assert config_entry.options[CONF_X10][0]["platform"] == "switch"
+
+
+async def test_add_x10_device_duplicate(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test adding a duplicate X10 device."""
+
+ x10_device = {"housecode": "a", "unitcode": 1, "platform": "switch"}
+
+ ws_client, _, _, _ = await async_mock_setup(
+ hass, hass_ws_client, config_options={CONF_X10: [x10_device]}
+ )
+ await ws_client.send_json(
+ {ID: 2, TYPE: "insteon/device/add_x10", "x10_device": x10_device}
+ )
+ msg = await ws_client.receive_json()
+ assert msg["error"]
+ assert msg["error"]["code"] == "duplicate"
+
+
+async def test_remove_device(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test removing an Insteon device."""
+ ws_client, _, _, _ = await async_mock_setup(hass, hass_ws_client)
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/device/remove",
+ "device_address": "11.22.33",
+ "remove_all_refs": True,
+ }
+ )
+ msg = await ws_client.receive_json()
+ assert msg["success"]
+
+
+async def test_remove_x10_device(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test removing an X10 device."""
+ ws_client, _, _, _ = await async_mock_setup(hass, hass_ws_client)
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/device/remove",
+ "device_address": "X10.A.01",
+ "remove_all_refs": True,
+ }
+ )
+ msg = await ws_client.receive_json()
+ assert msg["success"]
+
+
+async def test_remove_one_x10_device(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test one X10 device without removing others."""
+ x10_device = {"housecode": "a", "unitcode": 1, "platform": "light", "dim_steps": 22}
+ x10_devices = [
+ x10_device,
+ {"housecode": "a", "unitcode": 2, "platform": "switch"},
+ ]
+ ws_client, _, _, _ = await async_mock_setup(
+ hass, hass_ws_client, config_options={CONF_X10: x10_devices}
+ )
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/device/remove",
+ "device_address": "X10.A.01",
+ "remove_all_refs": True,
+ }
+ )
+ msg = await ws_client.receive_json()
+ assert msg["success"]
+ config_entry = hass.config_entries.async_get_entry("abcde12345")
+ assert len(config_entry.options[CONF_X10]) == 1
+ assert config_entry.options[CONF_X10][0]["housecode"] == "a"
+ assert config_entry.options[CONF_X10][0]["unitcode"] == 2
+
+
+async def test_remove_device_with_overload(
+ hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+ """Test removing an Insteon device that has a device overload."""
+ overload = {"address": "99.99.99", "cat": 1, "subcat": 3}
+ overloads = {CONF_OVERRIDE: [overload]}
+ ws_client, _, _, _ = await async_mock_setup(
+ hass, hass_ws_client, config_options=overloads
+ )
+ await ws_client.send_json(
+ {
+ ID: 2,
+ TYPE: "insteon/device/remove",
+ "device_address": "99.99.99",
+ "remove_all_refs": True,
+ }
+ )
+ msg = await ws_client.receive_json()
+ assert msg["success"]
+
+ config_entry = hass.config_entries.async_get_entry("abcde12345")
+ assert not config_entry.options.get(CONF_OVERRIDE)
diff --git a/tests/components/insteon/test_config_flow.py b/tests/components/insteon/test_config_flow.py
index 7cc0eefc0b558f6d606150e7fadf05b27b8dea37..4d3fb8154630cf2b1f47284b3565010f4a13c895 100644
--- a/tests/components/insteon/test_config_flow.py
+++ b/tests/components/insteon/test_config_flow.py
@@ -8,38 +8,14 @@ from voluptuous_serialize import convert
from homeassistant import config_entries
from homeassistant.components import dhcp, usb
from homeassistant.components.insteon.config_flow import (
- STEP_ADD_OVERRIDE,
- STEP_ADD_X10,
- STEP_CHANGE_HUB_CONFIG,
- STEP_CHANGE_PLM_CONFIG,
STEP_HUB_V1,
STEP_HUB_V2,
STEP_PLM,
STEP_PLM_MANUALLY,
- STEP_REMOVE_OVERRIDE,
- STEP_REMOVE_X10,
-)
-from homeassistant.components.insteon.const import (
- CONF_CAT,
- CONF_DIM_STEPS,
- CONF_HOUSECODE,
- CONF_HUB_VERSION,
- CONF_OVERRIDE,
- CONF_SUBCAT,
- CONF_UNITCODE,
- CONF_X10,
- DOMAIN,
)
+from homeassistant.components.insteon.const import CONF_HUB_VERSION, DOMAIN
from homeassistant.config_entries import ConfigEntryState
-from homeassistant.const import (
- CONF_ADDRESS,
- CONF_DEVICE,
- CONF_HOST,
- CONF_PASSWORD,
- CONF_PLATFORM,
- CONF_PORT,
- CONF_USERNAME,
-)
+from homeassistant.const import CONF_DEVICE, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@@ -52,11 +28,8 @@ 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
@@ -294,379 +267,6 @@ async def test_failed_connection_hub(hass: HomeAssistant) -> None:
assert result2["errors"] == {"base": "cannot_connect"}
-async def _options_init_form(hass, entry_id, step):
- """Run the init options form."""
- with patch(PATCH_ASYNC_SETUP_ENTRY, return_value=True):
- result = await hass.config_entries.options.async_init(entry_id)
-
- assert result["type"] is FlowResultType.MENU
- assert result["step_id"] == "init"
-
- return await hass.config_entries.options.async_configure(
- result["flow_id"],
- {"next_step_id": step},
- )
-
-
-async def _options_form(
- hass, flow_id, user_input, connection=mock_successful_connection
-):
- """Test an options form."""
- 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
-
-
-async def test_options_change_hub_config(hass: HomeAssistant) -> None:
- """Test changing Hub v2 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)
- assert result["type"] is 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"] is 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(
- 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)
-
- assert result["type"] is FlowResultType.CREATE_ENTRY
- assert config_entry.options == {}
- 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"] is FlowResultType.FORM
-
- assert result["type"] is 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(
- 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_ADD_OVERRIDE)
-
- user_input = {
- CONF_ADDRESS: "1a2b3c",
- CONF_CAT: "0x04",
- CONF_SUBCAT: "0xaa",
- }
- result, _ = await _options_form(hass, result["flow_id"], user_input)
-
- assert result["type"] is FlowResultType.CREATE_ENTRY
- assert len(config_entry.options[CONF_OVERRIDE]) == 1
- assert config_entry.options[CONF_OVERRIDE][0][CONF_ADDRESS] == "1A.2B.3C"
- assert config_entry.options[CONF_OVERRIDE][0][CONF_CAT] == 4
- assert config_entry.options[CONF_OVERRIDE][0][CONF_SUBCAT] == 170
-
- result2 = await _options_init_form(hass, config_entry.entry_id, STEP_ADD_OVERRIDE)
-
- user_input = {
- CONF_ADDRESS: "4d5e6f",
- CONF_CAT: "05",
- CONF_SUBCAT: "bb",
- }
- result3, _ = await _options_form(hass, result2["flow_id"], user_input)
-
- assert len(config_entry.options[CONF_OVERRIDE]) == 2
- assert config_entry.options[CONF_OVERRIDE][1][CONF_ADDRESS] == "4D.5E.6F"
- assert config_entry.options[CONF_OVERRIDE][1][CONF_CAT] == 5
- assert config_entry.options[CONF_OVERRIDE][1][CONF_SUBCAT] == 187
-
- # If result1 eq result2 the changes will not save
- assert result["data"] != result3["data"]
-
-
-async def test_options_remove_device_override(hass: HomeAssistant) -> None:
- """Test removing a device override."""
- config_entry = MockConfigEntry(
- domain=DOMAIN,
- entry_id="abcde12345",
- data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
- options={
- CONF_OVERRIDE: [
- {CONF_ADDRESS: "1A.2B.3C", CONF_CAT: 6, CONF_SUBCAT: 100},
- {CONF_ADDRESS: "4D.5E.6F", CONF_CAT: 7, CONF_SUBCAT: 200},
- ]
- },
- )
-
- config_entry.add_to_hass(hass)
- result = await _options_init_form(hass, config_entry.entry_id, STEP_REMOVE_OVERRIDE)
-
- user_input = {CONF_ADDRESS: "1A.2B.3C"}
- result, _ = await _options_form(hass, result["flow_id"], user_input)
-
- assert result["type"] is FlowResultType.CREATE_ENTRY
- assert len(config_entry.options[CONF_OVERRIDE]) == 1
-
-
-async def test_options_remove_device_override_with_x10(hass: HomeAssistant) -> None:
- """Test removing a device override when an X10 device is configured."""
- config_entry = MockConfigEntry(
- domain=DOMAIN,
- entry_id="abcde12345",
- data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
- options={
- CONF_OVERRIDE: [
- {CONF_ADDRESS: "1A.2B.3C", CONF_CAT: 6, CONF_SUBCAT: 100},
- {CONF_ADDRESS: "4D.5E.6F", CONF_CAT: 7, CONF_SUBCAT: 200},
- ],
- CONF_X10: [
- {
- CONF_HOUSECODE: "d",
- CONF_UNITCODE: 5,
- CONF_PLATFORM: "light",
- CONF_DIM_STEPS: 22,
- }
- ],
- },
- )
-
- config_entry.add_to_hass(hass)
- result = await _options_init_form(hass, config_entry.entry_id, STEP_REMOVE_OVERRIDE)
-
- user_input = {CONF_ADDRESS: "1A.2B.3C"}
- result, _ = await _options_form(hass, result["flow_id"], user_input)
-
- assert result["type"] is FlowResultType.CREATE_ENTRY
- assert len(config_entry.options[CONF_OVERRIDE]) == 1
- assert len(config_entry.options[CONF_X10]) == 1
-
-
-async def test_options_add_x10_device(hass: HomeAssistant) -> None:
- """Test adding an X10 device."""
- 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_ADD_X10)
-
- user_input = {
- CONF_HOUSECODE: "c",
- CONF_UNITCODE: 12,
- CONF_PLATFORM: "light",
- CONF_DIM_STEPS: 18,
- }
- result2, _ = await _options_form(hass, result["flow_id"], user_input)
-
- assert result2["type"] is FlowResultType.CREATE_ENTRY
- assert len(config_entry.options[CONF_X10]) == 1
- assert config_entry.options[CONF_X10][0][CONF_HOUSECODE] == "c"
- assert config_entry.options[CONF_X10][0][CONF_UNITCODE] == 12
- assert config_entry.options[CONF_X10][0][CONF_PLATFORM] == "light"
- assert config_entry.options[CONF_X10][0][CONF_DIM_STEPS] == 18
-
- result = await _options_init_form(hass, config_entry.entry_id, STEP_ADD_X10)
- user_input = {
- CONF_HOUSECODE: "d",
- CONF_UNITCODE: 10,
- CONF_PLATFORM: "binary_sensor",
- CONF_DIM_STEPS: 15,
- }
- result3, _ = await _options_form(hass, result["flow_id"], user_input)
-
- assert result3["type"] is FlowResultType.CREATE_ENTRY
- assert len(config_entry.options[CONF_X10]) == 2
- assert config_entry.options[CONF_X10][1][CONF_HOUSECODE] == "d"
- assert config_entry.options[CONF_X10][1][CONF_UNITCODE] == 10
- assert config_entry.options[CONF_X10][1][CONF_PLATFORM] == "binary_sensor"
- assert config_entry.options[CONF_X10][1][CONF_DIM_STEPS] == 15
-
- # If result2 eq result3 the changes will not save
- assert result2["data"] != result3["data"]
-
-
-async def test_options_remove_x10_device(hass: HomeAssistant) -> None:
- """Test removing an X10 device."""
- config_entry = MockConfigEntry(
- domain=DOMAIN,
- entry_id="abcde12345",
- data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
- options={
- CONF_X10: [
- {
- CONF_HOUSECODE: "C",
- CONF_UNITCODE: 4,
- CONF_PLATFORM: "light",
- CONF_DIM_STEPS: 18,
- },
- {
- CONF_HOUSECODE: "D",
- CONF_UNITCODE: 10,
- CONF_PLATFORM: "binary_sensor",
- CONF_DIM_STEPS: 15,
- },
- ]
- },
- )
-
- config_entry.add_to_hass(hass)
- result = await _options_init_form(hass, config_entry.entry_id, STEP_REMOVE_X10)
-
- user_input = {CONF_DEVICE: "Housecode: C, Unitcode: 4"}
- result, _ = await _options_form(hass, result["flow_id"], user_input)
-
- assert result["type"] is FlowResultType.CREATE_ENTRY
- assert len(config_entry.options[CONF_X10]) == 1
-
-
-async def test_options_remove_x10_device_with_override(hass: HomeAssistant) -> None:
- """Test removing an X10 device when a device override is configured."""
- config_entry = MockConfigEntry(
- domain=DOMAIN,
- entry_id="abcde12345",
- data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
- options={
- CONF_X10: [
- {
- CONF_HOUSECODE: "C",
- CONF_UNITCODE: 4,
- CONF_PLATFORM: "light",
- CONF_DIM_STEPS: 18,
- },
- {
- CONF_HOUSECODE: "D",
- CONF_UNITCODE: 10,
- CONF_PLATFORM: "binary_sensor",
- CONF_DIM_STEPS: 15,
- },
- ],
- CONF_OVERRIDE: [{CONF_ADDRESS: "1A.2B.3C", CONF_CAT: 1, CONF_SUBCAT: 18}],
- },
- )
-
- config_entry.add_to_hass(hass)
- result = await _options_init_form(hass, config_entry.entry_id, STEP_REMOVE_X10)
-
- user_input = {CONF_DEVICE: "Housecode: C, Unitcode: 4"}
- result, _ = await _options_form(hass, result["flow_id"], user_input)
-
- assert result["type"] is FlowResultType.CREATE_ENTRY
- assert len(config_entry.options[CONF_X10]) == 1
- assert len(config_entry.options[CONF_OVERRIDE]) == 1
-
-
-async def test_options_override_bad_data(hass: HomeAssistant) -> None:
- """Test for bad data in a device override."""
-
- 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_ADD_OVERRIDE)
-
- user_input = {
- CONF_ADDRESS: "zzzzzz",
- CONF_CAT: "bad",
- CONF_SUBCAT: "data",
- }
- result, _ = await _options_form(hass, result["flow_id"], user_input)
-
- assert result["type"] is FlowResultType.FORM
- assert result["errors"] == {"base": "input_error"}
-
-
async def test_discovery_via_usb(hass: HomeAssistant) -> None:
"""Test usb flow."""
discovery_info = usb.UsbServiceInfo(
diff --git a/tests/components/insteon/test_init.py b/tests/components/insteon/test_init.py
index a4e8da03345baeb977101bc2e727e8fe6bb73318..c5524ff19192e5e9c492d26aae099822152154ef 100644
--- a/tests/components/insteon/test_init.py
+++ b/tests/components/insteon/test_init.py
@@ -1,6 +1,5 @@
"""Test the init file for the Insteon component."""
-import asyncio
from unittest.mock import patch
import pytest
@@ -11,7 +10,7 @@ from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
-from .const import MOCK_USER_INPUT_PLM, PATCH_CONNECTION
+from .const import MOCK_USER_INPUT_PLM
from .mock_devices import MockDevices
from tests.common import MockConfigEntry
@@ -70,22 +69,24 @@ async def test_setup_entry_failed_connection(
async def test_import_frontend_dev_url(hass: HomeAssistant) -> None:
"""Test importing a dev_url config entry."""
- config = {}
- config[DOMAIN] = {CONF_DEV_PATH: "/some/path"}
+ config_entry = MockConfigEntry(
+ domain=DOMAIN, data=MOCK_USER_INPUT_PLM, options={CONF_DEV_PATH: "/some/path"}
+ )
+ config_entry.add_to_hass(hass)
with (
patch.object(insteon, "async_connect", new=mock_successful_connection),
- patch.object(insteon, "close_insteon_connection"),
+ patch.object(insteon, "async_close") as mock_close,
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.data[DOMAIN][CONF_DEV_PATH] == "/some/path"
+ hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
+ await hass.async_block_till_done()
+ assert insteon.devices.async_save.call_count == 1
+ assert mock_close.called