mirror of https://github.com/home-assistant/core
320 lines
9.8 KiB
Python
320 lines
9.8 KiB
Python
"""Test the config flow for the Insteon integration."""
|
|
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from voluptuous_serialize import convert
|
|
|
|
from homeassistant import config_entries
|
|
from homeassistant.components import dhcp, usb
|
|
from homeassistant.components.insteon.config_flow import (
|
|
STEP_HUB_V1,
|
|
STEP_HUB_V2,
|
|
STEP_PLM,
|
|
STEP_PLM_MANUALLY,
|
|
)
|
|
from homeassistant.components.insteon.const import CONF_HUB_VERSION, DOMAIN
|
|
from homeassistant.config_entries import ConfigEntryState, ConfigFlowResult
|
|
from homeassistant.const import CONF_DEVICE, CONF_HOST
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.data_entry_flow import FlowResultType
|
|
|
|
from .const import (
|
|
MOCK_DEVICE,
|
|
MOCK_USER_INPUT_HUB_V1,
|
|
MOCK_USER_INPUT_HUB_V2,
|
|
MOCK_USER_INPUT_PLM,
|
|
MOCK_USER_INPUT_PLM_MANUAL,
|
|
PATCH_ASYNC_SETUP_ENTRY,
|
|
PATCH_CONNECTION,
|
|
PATCH_USB_LIST,
|
|
)
|
|
|
|
from tests.common import MockConfigEntry
|
|
|
|
USB_PORTS = {"/dev/ttyUSB0": "/dev/ttyUSB0", MOCK_DEVICE: MOCK_DEVICE}
|
|
|
|
|
|
async def mock_successful_connection(*args, **kwargs):
|
|
"""Return a successful connection."""
|
|
return True
|
|
|
|
|
|
async def mock_usb_list(hass: HomeAssistant):
|
|
"""Return a mock list of USB devices."""
|
|
return USB_PORTS
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def patch_usb_list():
|
|
"""Only setup the lock and required base platforms to speed up tests."""
|
|
with patch(
|
|
PATCH_USB_LIST,
|
|
mock_usb_list,
|
|
):
|
|
yield
|
|
|
|
|
|
async def mock_failed_connection(*args, **kwargs):
|
|
"""Return a failed connection."""
|
|
raise ConnectionError("Connection failed")
|
|
|
|
|
|
async def _init_form(hass: HomeAssistant, modem_type: str) -> ConfigFlowResult:
|
|
"""Run the user form."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
|
|
return await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": modem_type},
|
|
)
|
|
|
|
|
|
async def _device_form(
|
|
hass: HomeAssistant,
|
|
flow_id: str,
|
|
connection: Callable[..., Any],
|
|
user_input: dict[str, Any] | None,
|
|
) -> tuple[ConfigFlowResult, AsyncMock]:
|
|
"""Test the PLM, Hub v1 or Hub v2 form."""
|
|
with (
|
|
patch(
|
|
PATCH_CONNECTION,
|
|
new=connection,
|
|
),
|
|
patch(
|
|
PATCH_ASYNC_SETUP_ENTRY,
|
|
return_value=True,
|
|
) as mock_setup_entry,
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(flow_id, user_input)
|
|
await hass.async_block_till_done()
|
|
return result, mock_setup_entry
|
|
|
|
|
|
async def test_form_select_modem(hass: HomeAssistant) -> None:
|
|
"""Test we get a modem form."""
|
|
|
|
result = await _init_form(hass, STEP_HUB_V2)
|
|
assert result["step_id"] == STEP_HUB_V2
|
|
assert result["type"] is FlowResultType.FORM
|
|
|
|
|
|
async def test_fail_on_existing(hass: HomeAssistant) -> None:
|
|
"""Test we fail if the integration is already configured."""
|
|
config_entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
entry_id="abcde12345",
|
|
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
|
|
options={},
|
|
)
|
|
config_entry.add_to_hass(hass)
|
|
assert config_entry.state is ConfigEntryState.NOT_LOADED
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
|
|
context={"source": config_entries.SOURCE_USER},
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "single_instance_allowed"
|
|
|
|
|
|
async def test_form_select_plm(hass: HomeAssistant) -> None:
|
|
"""Test we set up the PLM correctly."""
|
|
|
|
result = await _init_form(hass, STEP_PLM)
|
|
|
|
result2, mock_setup_entry = await _device_form(
|
|
hass, result["flow_id"], mock_successful_connection, MOCK_USER_INPUT_PLM
|
|
)
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["data"] == MOCK_USER_INPUT_PLM
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_form_select_plm_no_usb(hass: HomeAssistant) -> None:
|
|
"""Test we set up the PLM when no comm ports are found."""
|
|
|
|
temp_usb_list = dict(USB_PORTS)
|
|
USB_PORTS.clear()
|
|
result = await _init_form(hass, STEP_PLM)
|
|
|
|
result2, _ = await _device_form(
|
|
hass, result["flow_id"], mock_successful_connection, None
|
|
)
|
|
USB_PORTS.update(temp_usb_list)
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result2["step_id"] == STEP_PLM_MANUALLY
|
|
|
|
|
|
async def test_form_select_plm_manual(hass: HomeAssistant) -> None:
|
|
"""Test we set up the PLM correctly."""
|
|
|
|
result = await _init_form(hass, STEP_PLM)
|
|
|
|
result2, mock_setup_entry = await _device_form(
|
|
hass, result["flow_id"], mock_failed_connection, MOCK_USER_INPUT_PLM_MANUAL
|
|
)
|
|
|
|
result3, mock_setup_entry = await _device_form(
|
|
hass, result2["flow_id"], mock_successful_connection, MOCK_USER_INPUT_PLM
|
|
)
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result3["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result3["data"] == MOCK_USER_INPUT_PLM
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_form_select_hub_v1(hass: HomeAssistant) -> None:
|
|
"""Test we set up the Hub v1 correctly."""
|
|
|
|
result = await _init_form(hass, STEP_HUB_V1)
|
|
|
|
result2, mock_setup_entry = await _device_form(
|
|
hass, result["flow_id"], mock_successful_connection, MOCK_USER_INPUT_HUB_V1
|
|
)
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["data"] == {
|
|
**MOCK_USER_INPUT_HUB_V1,
|
|
CONF_HUB_VERSION: 1,
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_form_select_hub_v2(hass: HomeAssistant) -> None:
|
|
"""Test we set up the Hub v2 correctly."""
|
|
|
|
result = await _init_form(hass, STEP_HUB_V2)
|
|
|
|
result2, mock_setup_entry = await _device_form(
|
|
hass, result["flow_id"], mock_successful_connection, MOCK_USER_INPUT_HUB_V2
|
|
)
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["data"] == {
|
|
**MOCK_USER_INPUT_HUB_V2,
|
|
CONF_HUB_VERSION: 2,
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_form_discovery_dhcp(hass: HomeAssistant) -> None:
|
|
"""Test the discovery of the Hub via DHCP."""
|
|
discovery_info = dhcp.DhcpServiceInfo("1.2.3.4", "", "aabbccddeeff")
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=discovery_info
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
|
|
result2 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": STEP_HUB_V2},
|
|
)
|
|
assert result2["type"] is FlowResultType.FORM
|
|
schema = convert(result2["data_schema"])
|
|
found_host = False
|
|
for field in schema:
|
|
if field["name"] == CONF_HOST:
|
|
assert field["default"] == "1.2.3.4"
|
|
found_host = True
|
|
assert found_host
|
|
|
|
|
|
async def test_failed_connection_plm(hass: HomeAssistant) -> None:
|
|
"""Test a failed connection with the PLM."""
|
|
|
|
result = await _init_form(hass, STEP_PLM)
|
|
|
|
result2, _ = await _device_form(
|
|
hass, result["flow_id"], mock_failed_connection, MOCK_USER_INPUT_PLM
|
|
)
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result2["errors"] == {"base": "cannot_connect"}
|
|
|
|
|
|
async def test_failed_connection_plm_manually(hass: HomeAssistant) -> None:
|
|
"""Test a failed connection with the PLM."""
|
|
|
|
result = await _init_form(hass, STEP_PLM)
|
|
|
|
result2, _ = await _device_form(
|
|
hass, result["flow_id"], mock_successful_connection, MOCK_USER_INPUT_PLM_MANUAL
|
|
)
|
|
result3, _ = await _device_form(
|
|
hass, result["flow_id"], mock_failed_connection, MOCK_USER_INPUT_PLM
|
|
)
|
|
assert result3["type"] is FlowResultType.FORM
|
|
assert result3["errors"] == {"base": "cannot_connect"}
|
|
|
|
|
|
async def test_failed_connection_hub(hass: HomeAssistant) -> None:
|
|
"""Test a failed connection with a Hub."""
|
|
|
|
result = await _init_form(hass, STEP_HUB_V2)
|
|
|
|
result2, _ = await _device_form(
|
|
hass, result["flow_id"], mock_failed_connection, MOCK_USER_INPUT_HUB_V2
|
|
)
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result2["errors"] == {"base": "cannot_connect"}
|
|
|
|
|
|
async def test_discovery_via_usb(hass: HomeAssistant) -> None:
|
|
"""Test usb flow."""
|
|
discovery_info = usb.UsbServiceInfo(
|
|
device="/dev/ttyINSTEON",
|
|
pid="AAAA",
|
|
vid="AAAA",
|
|
serial_number="1234",
|
|
description="insteon radio",
|
|
manufacturer="test",
|
|
)
|
|
result = await hass.config_entries.flow.async_init(
|
|
"insteon", context={"source": config_entries.SOURCE_USB}, data=discovery_info
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm_usb"
|
|
|
|
with patch(PATCH_CONNECTION):
|
|
result2 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={}
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["data"] == {"device": "/dev/ttyINSTEON"}
|
|
|
|
|
|
async def test_discovery_via_usb_already_setup(hass: HomeAssistant) -> None:
|
|
"""Test usb flow -- already setup."""
|
|
|
|
MockConfigEntry(
|
|
domain=DOMAIN, data={CONF_DEVICE: {CONF_DEVICE: "/dev/ttyUSB1"}}
|
|
).add_to_hass(hass)
|
|
|
|
discovery_info = usb.UsbServiceInfo(
|
|
device="/dev/ttyINSTEON",
|
|
pid="AAAA",
|
|
vid="AAAA",
|
|
serial_number="1234",
|
|
description="insteon radio",
|
|
manufacturer="test",
|
|
)
|
|
result = await hass.config_entries.flow.async_init(
|
|
"insteon", context={"source": config_entries.SOURCE_USB}, data=discovery_info
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "single_instance_allowed"
|