evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Apply code review for insteon config flow (#39171)

Browse source 
* Move options import to async_setup_entry

* Add tests for insteon init

* Move common constants to const

* Clean up to adhear to standards

* Create mock insteon device manager

* Update for HA standards

* Use keys and align to config_flow steps

* Fix default port for hub v1

* Update doc string to represent function

* Remove dump print commands

* Add modem_type entry

* Simplify dict key test

* Setup platforms in async_setup_entry

* Black

* Black tests
This commit is contained in:
Tom Harris 2020-08-30 19:15:09 -04:00 • committed by GitHub
parent f7e6b060a7
commit 75153dd4a3
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 552 additions and 275 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

@ -394,7 +394,6 @@ omit =
homeassistant/components/ihc/*
homeassistant/components/imap/sensor.py
homeassistant/components/imap_email_content/sensor.py
homeassistant/components/insteon/__init__.py
homeassistant/components/insteon/binary_sensor.py
homeassistant/components/insteon/climate.py
homeassistant/components/insteon/const.py

123
homeassistant/components/insteon/__init__.py
View file

@ -6,7 +6,6 @@ from pyinsteon import async_close, async_connect, devices
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_PLATFORM, EVENT_HOMEASSISTANT_STOP
from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY
from homeassistant.exceptions import ConfigEntryNotReady
from .const import (
@ -30,10 +29,19 @@ from .utils import (
)
_LOGGER = logging.getLogger(__name__)
OPTIONS = "options"
async def async_id_unknown_devices(config_dir):
"""Send device ID commands to all unidentified devices."""
async def async_get_device_config(hass, config_entry):
"""Initiate the connection and services."""
# Make a copy of addresses due to edge case where the list of devices could change during status update
# Cannot be done concurrently due to issues with the underlying protocol.
for address in list(devices):
try:
await devices[address].async_status()
except AttributeError:
pass
await devices.async_load(id_devices=1)
for addr in devices:
device = devices[addr]
@ -52,45 +60,7 @@ async def async_id_unknown_devices(config_dir):
if not device.aldb.is_loaded or not flags:
await device.async_read_config()
await devices.async_save(workdir=config_dir)
async def async_setup_platforms(hass, config_entry):
"""Initiate the connection and services."""
tasks = [
hass.config_entries.async_forward_entry_setup(config_entry, component)
for component in INSTEON_COMPONENTS
]
await asyncio.gather(*tasks)
for address in devices:
device = devices[address]
platforms = get_device_platforms(device)
if ON_OFF_EVENTS in platforms:
add_on_off_event_device(hass, device)
_LOGGER.debug("Insteon device count: %s", len(devices))
register_new_device_callback(hass)
async_register_services(hass)
device_registry = await hass.helpers.device_registry.async_get_registry()
device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, str(devices.modem.address))},
manufacturer="Smart Home",
name=f"{devices.modem.description} {devices.modem.address}",
model=f"{devices.modem.model} (0x{devices.modem.cat:02x}, 0x{devices.modem.subcat:02x})",
sw_version=f"{devices.modem.firmware:02x} Engine Version: {devices.modem.engine_version}",
)
# Make a copy of addresses due to edge case where the list of devices could change during status update
# Cannot be done concurrently due to issues with the underlying protocol.
for address in list(devices):
try:
await devices[address].async_status()
except AttributeError:
pass
await async_id_unknown_devices(hass.config.config_dir)
await devices.async_save(workdir=hass.config.config_dir)
async def close_insteon_connection(*args):
@ -98,30 +68,22 @@ async def close_insteon_connection(*args):
await async_close()
async def async_import_config(hass, conf):
"""Set up all of the config imported from yaml."""
data, options = convert_yaml_to_config_flow(conf)
# Create a config entry with the connection data
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=data
)
# If this is the first time we ran, update the config options
if result["type"] == RESULT_TYPE_CREATE_ENTRY and options:
entry = result["result"]
hass.config_entries.async_update_entry(
entry=entry,
options=options,
)
return result
async def async_setup(hass, config):
"""Set up the Insteon platform."""
if DOMAIN not in config:
return True
conf = config[DOMAIN]
hass.async_create_task(async_import_config(hass, conf))
data, options = convert_yaml_to_config_flow(conf)
if options:
hass.data[DOMAIN] = {}
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
@ -141,6 +103,19 @@ async def async_setup_entry(hass, entry):
workdir=hass.config.config_dir, id_devices=0, load_modem_aldb=0
)
# If options existed in YAML and have not already been saved to the config entry
# add them now
if (
not entry.options
and entry.source == SOURCE_IMPORT
and hass.data.get(DOMAIN)
and hass.data[DOMAIN].get(OPTIONS)
):
hass.config_entries.async_update_entry(
entry=entry,
options=hass.data[DOMAIN][OPTIONS],
)
for device_override in entry.options.get(CONF_OVERRIDE, []):
# Override the device default capabilities for a specific address
address = device_override.get("address")
@ -163,5 +138,31 @@ async def async_setup_entry(hass, entry):
)
device = devices.add_x10_device(housecode, unitcode, x10_type, steps)
asyncio.create_task(async_setup_platforms(hass, entry))
for component in INSTEON_COMPONENTS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
for address in devices:
device = devices[address]
platforms = get_device_platforms(device)
if ON_OFF_EVENTS in platforms:
add_on_off_event_device(hass, device)
_LOGGER.debug("Insteon device count: %s", len(devices))
register_new_device_callback(hass)
async_register_services(hass)
device_registry = await hass.helpers.device_registry.async_get_registry()
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, str(devices.modem.address))},
manufacturer="Smart Home",
name=f"{devices.modem.description} {devices.modem.address}",
model=f"{devices.modem.model} (0x{devices.modem.cat:02x}, 0x{devices.modem.subcat:02x})",
sw_version=f"{devices.modem.firmware:02x} Engine Version: {devices.modem.engine_version}",
)
asyncio.create_task(async_get_device_config(hass, entry))
return True

10
homeassistant/components/insteon/config_flow.py
View file

@ -115,14 +115,10 @@ class InsteonFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return InsteonOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input=None):
"""For backward compatibility."""
return await self.async_step_init(user_input=user_input)
async def async_step_init(self, user_input=None):
"""Init the config flow."""
errors = {}
if self._async_current_entries():
return self.async_abort(reason="already_configured")
return self.async_abort(reason="single_instance_allowed")
if user_input is not None:
selection = user_input.get(MODEM_TYPE)
@ -134,7 +130,7 @@ class InsteonFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
modem_types = [PLM, HUB1, HUB2]
data_schema = vol.Schema({vol.Required(MODEM_TYPE): vol.In(modem_types)})
return self.async_show_form(
step_id="init", data_schema=data_schema, errors=errors
step_id="user", data_schema=data_schema, errors=errors
)
async def async_step_plm(self, user_input=None):
@ -177,7 +173,7 @@ class InsteonFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
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="already_configured")
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)

6
homeassistant/components/insteon/schemas.py
View file

@ -271,11 +271,13 @@ def build_plm_schema(device=vol.UNDEFINED):
def build_hub_schema(
hub_version,
host=vol.UNDEFINED,
port=PORT_HUB_V2,
port=vol.UNDEFINED,
username=vol.UNDEFINED,
password=vol.UNDEFINED,
):
"""Build the Hub v2 schema for config flow."""
"""Build the Hub schema for config flow."""
if port == vol.UNDEFINED:
port = PORT_HUB_V2 if hub_version == 2 else PORT_HUB_V1
schema = {
vol.Required(CONF_HOST, default=host): str,
vol.Required(CONF_PORT, default=port): int,

44
homeassistant/components/insteon/strings.json
View file

@ -1,48 +1,46 @@
{
"config": {
"step": {
"init": {
"user": {
"title": "Insteon",
"description": "Select the Insteon modem type.",
"data": {
"plm": "PowerLink Modem (PLM)",
"hubv1": "Hub Version 1 (Pre-2014)",
"hubv2": "Hub Version 2"
"modem_type": "Modem type."
}
},
"plm": {
"title": "Insteon PLM",
"description": "Configure the Insteon PowerLink Modem (PLM).",
"data": {
"device": "PLM device (i.e. /dev/ttyUSB0 or COM3)"
"device": "[%key:common::config_flow::data::usb_path%]"
}
},
"hub1": {
"hubv1": {
"title": "Insteon Hub Version 1",
"description": "Configure the Insteon Hub Version 1 (pre-2014).",
"data": {
"host": "Hub IP address",
"port": "IP port"
"host": "[%key:common::config_flow::data::ip%]",
"port": "[%key:common::config_flow::data::port%]"
}
},
"hub2": {
"hubv2": {
"title": "Insteon Hub Version 2",
"description": "Configure the Insteon Hub Version 2.",
"data": {
"host": "Hub IP address",
"username": "Username",
"password": "Password",
"port": "IP port"
"host": "[%key:common::config_flow::data::ip%]",
"port": "[%key:common::config_flow::data::port%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"cannot_connect": "Failed to connect to the Insteon modem, please try again.",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"select_single": "Select one option."
},
"abort": {
"cannot_connect": "Unable to connect to the Insteon modem",
"already_configured": "An Insteon modem connection is already configured"
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed"
}
},
"options": {
@ -62,10 +60,10 @@
"title": "Insteon",
"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.",
"data": {
"host": "New host name or IP address",
"username": "New username",
"password": "New password",
"port": "New port number"
"host": "[%key:common::config_flow::data::ip%]",
"port": "[%key:common::config_flow::data::port%]",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
},
"add_override": {
@ -103,13 +101,9 @@
}
},
"error": {
"cannot_connect": "Failed to connect to the Insteon modem, please try again.",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"select_single": "Select one option.",
"input_error": "Invalid entries, please check your values."
},
"abort": {
"cannot_connect": "Unable to connect to the Insteon modem",
"already_configured": "An Insteon modem connection is already configured"
}
}
}

51
homeassistant/components/insteon/translations/en.json
View file

@ -1,57 +1,48 @@
{
"config": {
"abort": {
"already_configured": "An Insteon modem connection is already configured",
"cannot_connect": "Unable to connect to the Insteon modem"
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed"
},
"error": {
"cannot_connect": "Failed to connect to the Insteon modem, please try again.",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"select_single": "Select one option."
},
"step": {
"hub1": {
"hubv1": {
"data": {
"host": "Hub IP address",
"port": "IP port"
"host": "[%key:common::config_flow::data::ip%]",
"port": "[%key:common::config_flow::data::port%]"
},
"description": "Configure the Insteon Hub Version 1 (pre-2014).",
"title": "Insteon Hub Version 1"
},
"hub2": {
"hubv2": {
"data": {
"host": "Hub IP address",
"password": "Password",
"port": "IP port",
"username": "Username"
"host": "[%key:common::config_flow::data::ip%]",
"password": "[%key:common::config_flow::data::password%]",
"port": "[%key:common::config_flow::data::port%]",
"username": "[%key:common::config_flow::data::username%]"
},
"description": "Configure the Insteon Hub Version 2.",
"title": "Insteon Hub Version 2"
},
"init": {
"data": {
"hubv1": "Hub Version 1 (Pre-2014)",
"hubv2": "Hub Version 2",
"plm": "PowerLink Modem (PLM)"
},
"description": "Select the Insteon modem type.",
"title": "Insteon"
},
"plm": {
"data": {
"device": "PLM device (i.e. /dev/ttyUSB0 or COM3)"
"device": "[%key:common::config_flow::data::usb_path%]"
},
"description": "Configure the Insteon PowerLink Modem (PLM).",
"title": "Insteon PLM"
},
"user": {
"description": "Select the Insteon modem type.",
"title": "Insteon"
}
}
},
"options": {
"abort": {
"already_configured": "An Insteon modem connection is already configured",
"cannot_connect": "Unable to connect to the Insteon modem"
},
"error": {
"cannot_connect": "Failed to connect to the Insteon modem, please try again.",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"input_error": "Invalid entries, please check your values.",
"select_single": "Select one option."
},
@ -77,10 +68,10 @@
},
"change_hub_config": {
"data": {
"host": "New host name or IP address",
"password": "New password",
"port": "New port number",
"username": "New username"
"host": "[%key:common::config_flow::data::ip%]",
"password": "[%key:common::config_flow::data::password%]",
"port": "[%key:common::config_flow::data::port%]",
"username": "[%key:common::config_flow::data::username%]"
},
"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.",
"title": "Insteon"

100
tests/components/insteon/const.py Normal file
View file

@ -0,0 +1,100 @@
"""Constants used for Insteon test cases."""
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 (
CONF_ADDRESS,
CONF_DEVICE,
CONF_HOST,
CONF_PASSWORD,
CONF_PLATFORM,
CONF_PORT,
CONF_USERNAME,
)
MOCK_HOSTNAME = "1.1.1.1"
MOCK_DEVICE = "/dev/ttyUSB55"
MOCK_USERNAME = "test-username"
MOCK_PASSWORD = "test-password"
MOCK_PORT = 4567
MOCK_ADDRESS = "1a2b3c"
MOCK_CAT = 0x02
MOCK_SUBCAT = 0x1A
MOCK_HOUSECODE = "c"
MOCK_UNITCODE_1 = 1
MOCK_UNITCODE_2 = 2
MOCK_X10_PLATFORM_1 = X10_PLATFORMS[0]
MOCK_X10_PLATFORM_2 = X10_PLATFORMS[2]
MOCK_X10_STEPS = 10
MOCK_USER_INPUT_PLM = {
CONF_DEVICE: MOCK_DEVICE,
}
MOCK_USER_INPUT_HUB_V2 = {
CONF_HOST: MOCK_HOSTNAME,
CONF_USERNAME: MOCK_USERNAME,
CONF_PASSWORD: MOCK_PASSWORD,
CONF_PORT: MOCK_PORT,
}
MOCK_USER_INPUT_HUB_V1 = {
CONF_HOST: MOCK_HOSTNAME,
CONF_PORT: MOCK_PORT,
}
MOCK_DEVICE_OVERRIDE_CONFIG = {
CONF_ADDRESS: MOCK_ADDRESS,
CONF_CAT: MOCK_CAT,
CONF_SUBCAT: MOCK_SUBCAT,
}
MOCK_X10_CONFIG_1 = {
CONF_HOUSECODE: MOCK_HOUSECODE,
CONF_UNITCODE: MOCK_UNITCODE_1,
CONF_PLATFORM: MOCK_X10_PLATFORM_1,
CONF_DIM_STEPS: MOCK_X10_STEPS,
}
MOCK_X10_CONFIG_2 = {
CONF_HOUSECODE: MOCK_HOUSECODE,
CONF_UNITCODE: MOCK_UNITCODE_2,
CONF_PLATFORM: MOCK_X10_PLATFORM_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_ASYNC_SETUP = "homeassistant.components.insteon.async_setup"
PATCH_ASYNC_SETUP_ENTRY = "homeassistant.components.insteon.async_setup_entry"

69
tests/components/insteon/mock_devices.py Normal file
View file

@ -0,0 +1,69 @@
"""Mock devices object to test Insteon."""
import logging
from pyinsteon.address import Address
from pyinsteon.device_types import (
GeneralController_MiniRemote_4,
Hub,
SwitchedLightingControl_SwitchLinc,
)
from tests.async_mock import AsyncMock, MagicMock
_LOGGER = logging.getLogger(__name__)
class MockSwitchLinc(SwitchedLightingControl_SwitchLinc):
"""Mock SwitchLinc device."""
@property
def operating_flags(self):
"""Return no operating flags to force properties to be checked."""
return {}
class MockDevices:
"""Mock devices class."""
def __init__(self, connected=True):
"""Init the MockDevices class."""
self._devices = {}
self.modem = None
self._connected = connected
self.async_save = AsyncMock()
self.add_x10_device = MagicMock()
self.set_id = MagicMock()
def __getitem__(self, address):
"""Return a a device from the device address."""
return self._devices.get(address)
def __iter__(self):
"""Return an iterator of device addresses."""
yield from self._devices
def __len__(self):
"""Return the number of devices."""
return len(self._devices)
def get(self, address):
"""Return a device from an address or None if not found."""
return self._devices.get(Address(address))
async def async_load(self, *args, **kwargs):
"""Load the mock devices."""
if self._connected:
addr0 = Address("AA.AA.AA")
addr1 = Address("11.11.11")
addr2 = Address("22.22.22")
addr3 = Address("33.33.33")
self._devices[addr0] = Hub(addr0)
self._devices[addr1] = MockSwitchLinc(addr1, 0x02, 0x00)
self._devices[addr2] = GeneralController_MiniRemote_4(addr2, 0x00, 0x00)
self._devices[addr3] = SwitchedLightingControl_SwitchLinc(addr3, 0x02, 0x00)
for device in [self._devices[addr] for addr in [addr1, addr2, addr3]]:
device.async_read_config = AsyncMock()
for device in [self._devices[addr] for addr in [addr2, addr3]]:
device.async_status = AsyncMock()
self._devices[addr1].async_status = AsyncMock(side_effect=AttributeError)
self.modem = self._devices[addr0]

196
tests/components/insteon/test_config_flow.py
View file

@ -1,7 +1,6 @@
"""Test the config flow for the Insteon integration."""
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.insteon import async_import_config
from homeassistant.components.insteon.config_flow import (
HUB1,
HUB2,
@ -24,7 +23,6 @@ from homeassistant.components.insteon.const import (
CONF_UNITCODE,
CONF_X10,
DOMAIN,
X10_PLATFORMS,
)
from homeassistant.const import (
CONF_ADDRESS,
@ -37,79 +35,24 @@ from homeassistant.const import (
)
from homeassistant.helpers.typing import HomeAssistantType
from .const import (
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,
)
from tests.async_mock import patch
from tests.common import MockConfigEntry
MOCK_HOSTNAME = "1.1.1.1"
MOCK_DEVICE = "/dev/ttyUSB55"
MOCK_USERNAME = "test-username"
MOCK_PASSWORD = "test-password"
MOCK_PORT = 4567
MOCK_ADDRESS = "1a2b3c"
MOCK_CAT = 0x02
MOCK_SUBCAT = 0x1A
MOCK_HOUSECODE = "c"
MOCK_UNITCODE = 6
MOCK_X10_PLATFORM = X10_PLATFORMS[2]
MOCK_X10_STEPS = 10
MOCK_USER_INPUT_PLM = {
CONF_DEVICE: MOCK_DEVICE,
}
MOCK_USER_INPUT_HUB_V2 = {
CONF_HOST: MOCK_HOSTNAME,
CONF_USERNAME: MOCK_USERNAME,
CONF_PASSWORD: MOCK_PASSWORD,
CONF_PORT: MOCK_PORT,
}
MOCK_USER_INPUT_HUB_V1 = {
CONF_HOST: MOCK_HOSTNAME,
CONF_PORT: MOCK_PORT,
}
MOCK_DEVICE_OVERRIDE_CONFIG = {
CONF_ADDRESS: MOCK_ADDRESS,
CONF_CAT: MOCK_CAT,
CONF_SUBCAT: MOCK_SUBCAT,
}
MOCK_X10_CONFIG = {
CONF_HOUSECODE: MOCK_HOUSECODE,
CONF_UNITCODE: MOCK_UNITCODE,
CONF_PLATFORM: MOCK_X10_PLATFORM,
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]
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]
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]
PATCH_CONNECTION = "homeassistant.components.insteon.config_flow.async_connect"
PATCH_ASYNC_SETUP = "homeassistant.components.insteon.async_setup"
PATCH_ASYNC_SETUP_ENTRY = "homeassistant.components.insteon.async_setup_entry"
async def mock_successful_connection(*args, **kwargs):
"""Return a successful connection."""
@ -122,7 +65,7 @@ async def mock_failed_connection(*args, **kwargs):
async def _init_form(hass, modem_type):
"""Run the init form."""
"""Run the user form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@ -173,7 +116,7 @@ async def test_fail_on_existing(hass: HomeAssistantType):
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
assert result["reason"] == "single_instance_allowed"
async def test_form_select_plm(hass: HomeAssistantType):
@ -259,7 +202,9 @@ async def _import_config(hass, config):
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 async_import_config(hass, config)
return await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
)
async def test_import_plm(hass: HomeAssistantType):
@ -271,71 +216,31 @@ async def test_import_plm(hass: HomeAssistantType):
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_USER_INPUT_PLM
assert entry.data == MOCK_IMPORT_CONFIG_PLM
async def test_import_plm_full(hass: HomeAssistantType):
"""Test importing a full PLM config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_FULL_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_USER_INPUT_PLM
assert entry.options[CONF_OVERRIDE][0][CONF_ADDRESS] == "1A.2B.3C"
assert entry.options[CONF_OVERRIDE][0][CONF_CAT] == MOCK_CAT
assert entry.options[CONF_OVERRIDE][0][CONF_SUBCAT] == MOCK_SUBCAT
assert entry.options[CONF_X10][0][CONF_HOUSECODE] == MOCK_HOUSECODE
assert entry.options[CONF_X10][0][CONF_UNITCODE] == MOCK_UNITCODE
assert entry.options[CONF_X10][0][CONF_PLATFORM] == MOCK_X10_PLATFORM
assert entry.options[CONF_X10][0][CONF_DIM_STEPS] == MOCK_X10_STEPS
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"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
async def test_import_full_hub_v1(hass: HomeAssistantType):
"""Test importing a full Hub v1 config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_FULL_CONFIG_HUB_V1)
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] == MOCK_PORT
assert entry.data[CONF_HUB_VERSION] == 1
assert CONF_USERNAME not in entry.data
assert CONF_PASSWORD not in entry.data
assert CONF_OVERRIDE not in entry.data
assert CONF_X10 not in entry.data
assert entry.options[CONF_OVERRIDE][0][CONF_ADDRESS] == "1A.2B.3C"
assert entry.options[CONF_X10][0][CONF_HOUSECODE] == MOCK_HOUSECODE
async def test_import_full_hub_v2(hass: HomeAssistantType):
"""Test importing a full Hub v2 config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_FULL_CONFIG_HUB_V2)
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] == MOCK_PORT
assert entry.data[CONF_USERNAME] == MOCK_USERNAME
assert entry.data[CONF_PASSWORD] == MOCK_PASSWORD
assert entry.data[CONF_HUB_VERSION] == 2
assert CONF_OVERRIDE not in entry.data
assert CONF_X10 not in entry.data
assert entry.options[CONF_OVERRIDE][0][CONF_ADDRESS] == "1A.2B.3C"
assert entry.options[CONF_X10][0][CONF_HOUSECODE] == MOCK_HOUSECODE
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
{step: True},
)
return result2
async def test_import_min_hub_v2(hass: HomeAssistantType):
"""Test importing a minimum Hub v2 config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_MINIMUM_HUB_V2)
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)
@ -351,7 +256,9 @@ async def test_import_min_hub_v1(hass: HomeAssistantType):
"""Test importing a minimum Hub v1 config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_MINIMUM_HUB_V1)
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)
@ -372,9 +279,11 @@ async def test_import_existing(hass: HomeAssistantType):
config_entry.add_to_hass(hass)
assert config_entry.state == config_entries.ENTRY_STATE_NOT_LOADED
result = await _import_config(hass, MOCK_IMPORT_MINIMUM_HUB_V2)
result = await _import_config(
hass, {**MOCK_IMPORT_MINIMUM_HUB_V2, CONF_PORT: 25105, CONF_HUB_VERSION: 2}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
assert result["reason"] == "single_instance_allowed"
async def test_import_failed_connection(hass: HomeAssistantType):
@ -384,27 +293,16 @@ async def test_import_failed_connection(hass: HomeAssistantType):
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 async_import_config(hass, MOCK_IMPORT_MINIMUM_HUB_V2)
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.RESULT_TYPE_ABORT
assert result["reason"] == "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"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
{step: True},
)
return result2
async def _options_form(hass, flow_id, user_input):
"""Test an options form."""

227
tests/components/insteon/test_init.py Normal file
View file

@ -0,0 +1,227 @@
"""Test the init file for the Insteon component."""
import asyncio
import logging
from pyinsteon.address import Address
from homeassistant.components import insteon
from homeassistant.components.insteon.const import (
CONF_CAT,
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.helpers.typing import HomeAssistantType
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 .mock_devices import MockDevices
from tests.async_mock import patch
from tests.common import MockConfigEntry
_LOGGER = logging.getLogger(__name__)
async def mock_successful_connection(*args, **kwargs):
"""Return a successful connection."""
return True
async def mock_failed_connection(*args, **kwargs):
"""Return a failed connection."""
raise ConnectionError("Connection failed")
async def test_setup_entry(hass: HomeAssistantType):
"""Test setting up the entry."""
config_entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_INPUT_PLM)
config_entry.add_to_hass(hass)
with patch.object(
insteon, "async_connect", new=mock_successful_connection
), patch.object(insteon, "async_close") as mock_close, patch.object(
insteon, "devices", new=MockDevices()
):
assert await async_setup_component(
hass,
insteon.DOMAIN,
{},
)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
# pylint: disable=no-member
assert insteon.devices.async_save.call_count == 1
assert mock_close.called
async def test_import_plm(hass: HomeAssistantType):
"""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: HomeAssistantType):
"""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: HomeAssistantType):
"""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: HomeAssistantType):
"""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: HomeAssistantType):
"""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: HomeAssistantType, caplog):
"""Test setting up the entry with a failed connection."""
config_entry = MockConfigEntry(domain=DOMAIN, data=MOCK_USER_INPUT_PLM)
config_entry.add_to_hass(hass)
with patch.object(
insteon, "async_connect", new=mock_failed_connection
), patch.object(insteon, "devices", new=MockDevices(connected=False)):
assert await async_setup_component(
hass,
insteon.DOMAIN,
{},
)
assert "Could not connect to Insteon modem" in caplog.text