mirror of https://github.com/home-assistant/core
944 lines
32 KiB
Python
944 lines
32 KiB
Python
"""Test the Yeelight config flow."""
|
|
|
|
from ipaddress import ip_address
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from homeassistant import config_entries
|
|
from homeassistant.components import dhcp, ssdp, zeroconf
|
|
from homeassistant.components.yeelight.config_flow import (
|
|
MODEL_UNKNOWN,
|
|
CannotConnect,
|
|
YeelightConfigFlow,
|
|
)
|
|
from homeassistant.components.yeelight.const import (
|
|
CONF_DETECTED_MODEL,
|
|
CONF_MODE_MUSIC,
|
|
CONF_NIGHTLIGHT_SWITCH,
|
|
CONF_NIGHTLIGHT_SWITCH_TYPE,
|
|
CONF_SAVE_ON_CHANGE,
|
|
CONF_TRANSITION,
|
|
DEFAULT_MODE_MUSIC,
|
|
DEFAULT_NAME,
|
|
DEFAULT_NIGHTLIGHT_SWITCH,
|
|
DEFAULT_SAVE_ON_CHANGE,
|
|
DEFAULT_TRANSITION,
|
|
DOMAIN,
|
|
NIGHTLIGHT_SWITCH_TYPE_LIGHT,
|
|
)
|
|
from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_ID, CONF_MODEL, CONF_NAME
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.data_entry_flow import FlowResultType
|
|
|
|
from . import (
|
|
CAPABILITIES,
|
|
ID,
|
|
IP_ADDRESS,
|
|
MODEL,
|
|
MODULE,
|
|
MODULE_CONFIG_FLOW,
|
|
NAME,
|
|
UNIQUE_FRIENDLY_NAME,
|
|
ZEROCONF_DATA,
|
|
_mocked_bulb,
|
|
_patch_discovery,
|
|
_patch_discovery_interval,
|
|
_patch_discovery_timeout,
|
|
)
|
|
|
|
from tests.common import MockConfigEntry
|
|
|
|
DEFAULT_CONFIG = {
|
|
CONF_MODEL: "",
|
|
CONF_TRANSITION: DEFAULT_TRANSITION,
|
|
CONF_MODE_MUSIC: DEFAULT_MODE_MUSIC,
|
|
CONF_SAVE_ON_CHANGE: DEFAULT_SAVE_ON_CHANGE,
|
|
CONF_NIGHTLIGHT_SWITCH: DEFAULT_NIGHTLIGHT_SWITCH,
|
|
}
|
|
|
|
SSDP_INFO = ssdp.SsdpServiceInfo(
|
|
ssdp_usn="mock_usn",
|
|
ssdp_st="mock_st",
|
|
upnp={},
|
|
ssdp_headers=CAPABILITIES,
|
|
)
|
|
|
|
|
|
async def test_discovery(hass: HomeAssistant) -> None:
|
|
"""Test setting up discovery."""
|
|
with _patch_discovery(), _patch_discovery_interval():
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
assert not result["errors"]
|
|
|
|
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result2["step_id"] == "pick_device"
|
|
assert not result2["errors"]
|
|
|
|
# test we can try again
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
assert not result["errors"]
|
|
|
|
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result2["step_id"] == "pick_device"
|
|
assert not result2["errors"]
|
|
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE}.async_setup", return_value=True) as mock_setup,
|
|
patch(f"{MODULE}.async_setup_entry", return_value=True) as mock_setup_entry,
|
|
):
|
|
result3 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], {CONF_DEVICE: ID}
|
|
)
|
|
assert result3["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result3["title"] == UNIQUE_FRIENDLY_NAME
|
|
assert result3["data"] == {CONF_ID: ID, CONF_HOST: IP_ADDRESS, CONF_MODEL: MODEL}
|
|
await hass.async_block_till_done()
|
|
mock_setup.assert_called_once()
|
|
mock_setup_entry.assert_called_once()
|
|
|
|
# ignore configured devices
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
assert not result["errors"]
|
|
|
|
with _patch_discovery(), _patch_discovery_interval():
|
|
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
|
assert result2["type"] is FlowResultType.ABORT
|
|
assert result2["reason"] == "no_devices_found"
|
|
|
|
|
|
async def test_discovery_with_existing_device_present(hass: HomeAssistant) -> None:
|
|
"""Test setting up discovery."""
|
|
config_entry = MockConfigEntry(
|
|
domain=DOMAIN, data={CONF_ID: "0x000000000099999", CONF_HOST: "4.4.4.4"}
|
|
)
|
|
config_entry.add_to_hass(hass)
|
|
alternate_bulb = _mocked_bulb()
|
|
alternate_bulb.capabilities["id"] = "0x000000000099999"
|
|
alternate_bulb.capabilities["location"] = "yeelight://4.4.4.4"
|
|
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_timeout(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE}.AsyncBulb", return_value=alternate_bulb),
|
|
):
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
await hass.async_block_till_done()
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
assert not result["errors"]
|
|
|
|
with _patch_discovery(), _patch_discovery_timeout(), _patch_discovery_interval():
|
|
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
|
await hass.async_block_till_done()
|
|
await hass.async_block_till_done()
|
|
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result2["step_id"] == "pick_device"
|
|
assert not result2["errors"]
|
|
|
|
# Now abort and make sure we can start over
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
assert not result["errors"]
|
|
|
|
with _patch_discovery(), _patch_discovery_interval():
|
|
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result2["step_id"] == "pick_device"
|
|
assert not result2["errors"]
|
|
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE}.AsyncBulb", return_value=_mocked_bulb()),
|
|
):
|
|
result3 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], {CONF_DEVICE: ID}
|
|
)
|
|
assert result3["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result3["title"] == UNIQUE_FRIENDLY_NAME
|
|
assert result3["data"] == {
|
|
CONF_ID: ID,
|
|
CONF_HOST: IP_ADDRESS,
|
|
CONF_MODEL: MODEL,
|
|
}
|
|
await hass.async_block_till_done()
|
|
await hass.async_block_till_done()
|
|
|
|
# ignore configured devices
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
assert not result["errors"]
|
|
|
|
with _patch_discovery(), _patch_discovery_interval():
|
|
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
|
assert result2["type"] is FlowResultType.ABORT
|
|
assert result2["reason"] == "no_devices_found"
|
|
|
|
|
|
async def test_discovery_no_device(hass: HomeAssistant) -> None:
|
|
"""Test discovery without device."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
|
|
with (
|
|
_patch_discovery(no_device=True),
|
|
_patch_discovery_timeout(),
|
|
_patch_discovery_interval(),
|
|
):
|
|
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
|
|
|
assert result2["type"] is FlowResultType.ABORT
|
|
assert result2["reason"] == "no_devices_found"
|
|
|
|
|
|
async def test_import(hass: HomeAssistant) -> None:
|
|
"""Test import from yaml."""
|
|
config = {
|
|
CONF_NAME: DEFAULT_NAME,
|
|
CONF_HOST: IP_ADDRESS,
|
|
CONF_TRANSITION: DEFAULT_TRANSITION,
|
|
CONF_MODE_MUSIC: DEFAULT_MODE_MUSIC,
|
|
CONF_SAVE_ON_CHANGE: DEFAULT_SAVE_ON_CHANGE,
|
|
CONF_NIGHTLIGHT_SWITCH_TYPE: NIGHTLIGHT_SWITCH_TYPE_LIGHT,
|
|
}
|
|
|
|
# Cannot connect
|
|
mocked_bulb = _mocked_bulb(cannot_connect=True)
|
|
with (
|
|
_patch_discovery(no_device=True),
|
|
_patch_discovery_timeout(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "cannot_connect"
|
|
|
|
# Success
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
patch(f"{MODULE}.async_setup", return_value=True) as mock_setup,
|
|
patch(f"{MODULE}.async_setup_entry", return_value=True) as mock_setup_entry,
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
|
|
)
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == DEFAULT_NAME
|
|
assert result["data"] == {
|
|
CONF_NAME: DEFAULT_NAME,
|
|
CONF_HOST: IP_ADDRESS,
|
|
CONF_TRANSITION: DEFAULT_TRANSITION,
|
|
CONF_MODE_MUSIC: DEFAULT_MODE_MUSIC,
|
|
CONF_SAVE_ON_CHANGE: DEFAULT_SAVE_ON_CHANGE,
|
|
CONF_NIGHTLIGHT_SWITCH: True,
|
|
}
|
|
await hass.async_block_till_done()
|
|
mock_setup.assert_called_once()
|
|
mock_setup_entry.assert_called_once()
|
|
|
|
# Duplicate
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
|
|
|
|
async def test_manual(hass: HomeAssistant) -> None:
|
|
"""Test manually setup."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
assert not result["errors"]
|
|
|
|
# Cannot connect (timeout)
|
|
mocked_bulb = _mocked_bulb(cannot_connect=True)
|
|
with (
|
|
_patch_discovery(no_device=True),
|
|
_patch_discovery_timeout(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result2 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], {CONF_HOST: IP_ADDRESS}
|
|
)
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result2["step_id"] == "user"
|
|
assert result2["errors"] == {"base": "cannot_connect"}
|
|
|
|
# Cannot connect (error)
|
|
with (
|
|
_patch_discovery(no_device=True),
|
|
_patch_discovery_timeout(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result3 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], {CONF_HOST: IP_ADDRESS}
|
|
)
|
|
assert result3["errors"] == {"base": "cannot_connect"}
|
|
|
|
# Success
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_timeout(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
patch(f"{MODULE}.async_setup", return_value=True),
|
|
patch(f"{MODULE}.async_setup_entry", return_value=True),
|
|
):
|
|
result4 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], {CONF_HOST: IP_ADDRESS}
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result4["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result4["title"] == "Color 0x15243f"
|
|
assert result4["data"] == {
|
|
CONF_HOST: IP_ADDRESS,
|
|
CONF_ID: "0x000000000015243f",
|
|
CONF_MODEL: MODEL,
|
|
}
|
|
|
|
# Duplicate
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(no_device=True),
|
|
_patch_discovery_timeout(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result2 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], {CONF_HOST: IP_ADDRESS}
|
|
)
|
|
assert result2["type"] is FlowResultType.ABORT
|
|
assert result2["reason"] == "already_configured"
|
|
|
|
|
|
async def test_options(hass: HomeAssistant) -> None:
|
|
"""Test options flow."""
|
|
config_entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
data={CONF_HOST: IP_ADDRESS, CONF_NAME: NAME, CONF_DETECTED_MODEL: MODEL},
|
|
)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with _patch_discovery(), patch(f"{MODULE}.AsyncBulb", return_value=mocked_bulb):
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
config = {
|
|
CONF_NAME: NAME,
|
|
CONF_MODEL: MODEL,
|
|
CONF_TRANSITION: DEFAULT_TRANSITION,
|
|
CONF_MODE_MUSIC: DEFAULT_MODE_MUSIC,
|
|
CONF_SAVE_ON_CHANGE: DEFAULT_SAVE_ON_CHANGE,
|
|
CONF_NIGHTLIGHT_SWITCH: DEFAULT_NIGHTLIGHT_SWITCH,
|
|
}
|
|
assert config_entry.options == config
|
|
assert hass.states.get(f"light.{NAME}_nightlight") is None
|
|
|
|
result = await hass.config_entries.options.async_init(config_entry.entry_id)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "init"
|
|
|
|
config[CONF_NIGHTLIGHT_SWITCH] = True
|
|
user_input = {**config}
|
|
user_input.pop(CONF_NAME)
|
|
user_input.pop(CONF_MODEL)
|
|
with _patch_discovery(), patch(f"{MODULE}.AsyncBulb", return_value=mocked_bulb):
|
|
result2 = await hass.config_entries.options.async_configure(
|
|
result["flow_id"], user_input
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["data"] == config
|
|
assert result2["data"] == config_entry.options
|
|
assert hass.states.get(f"light.{NAME}_nightlight") is not None
|
|
|
|
|
|
async def test_options_unknown_model(hass: HomeAssistant) -> None:
|
|
"""Test options flow with an unknown model."""
|
|
config_entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
data={CONF_HOST: IP_ADDRESS, CONF_NAME: NAME, CONF_DETECTED_MODEL: "not_in_db"},
|
|
)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with _patch_discovery(), patch(f"{MODULE}.AsyncBulb", return_value=mocked_bulb):
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
config = {
|
|
CONF_NAME: NAME,
|
|
CONF_MODEL: "not_in_db",
|
|
CONF_TRANSITION: DEFAULT_TRANSITION,
|
|
CONF_MODE_MUSIC: DEFAULT_MODE_MUSIC,
|
|
CONF_SAVE_ON_CHANGE: DEFAULT_SAVE_ON_CHANGE,
|
|
CONF_NIGHTLIGHT_SWITCH: DEFAULT_NIGHTLIGHT_SWITCH,
|
|
}
|
|
assert config_entry.options == config
|
|
assert hass.states.get(f"light.{NAME}_nightlight") is None
|
|
|
|
result = await hass.config_entries.options.async_init(config_entry.entry_id)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "init"
|
|
|
|
config[CONF_NIGHTLIGHT_SWITCH] = True
|
|
user_input = {**config}
|
|
user_input.pop(CONF_NAME)
|
|
with _patch_discovery(), patch(f"{MODULE}.AsyncBulb", return_value=mocked_bulb):
|
|
result2 = await hass.config_entries.options.async_configure(
|
|
result["flow_id"], user_input
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["data"] == config
|
|
assert result2["data"] == config_entry.options
|
|
assert hass.states.get(f"light.{NAME}_nightlight") is not None
|
|
|
|
|
|
async def test_manual_no_capabilities(hass: HomeAssistant) -> None:
|
|
"""Test manually setup without successful get_capabilities."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "user"
|
|
assert not result["errors"]
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(no_device=True),
|
|
_patch_discovery_timeout(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
patch(
|
|
f"{MODULE}.async_setup",
|
|
return_value=True,
|
|
),
|
|
patch(
|
|
f"{MODULE}.async_setup_entry",
|
|
return_value=True,
|
|
),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], {CONF_HOST: IP_ADDRESS}
|
|
)
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"] == {
|
|
CONF_HOST: IP_ADDRESS,
|
|
CONF_ID: None,
|
|
CONF_MODEL: MODEL_UNKNOWN,
|
|
}
|
|
|
|
|
|
async def test_discovered_by_homekit_and_dhcp(hass: HomeAssistant) -> None:
|
|
"""Test we get the form with homekit and abort for dhcp source when we get both."""
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": config_entries.SOURCE_HOMEKIT},
|
|
data=zeroconf.ZeroconfServiceInfo(
|
|
ip_address=ip_address(IP_ADDRESS),
|
|
ip_addresses=[ip_address(IP_ADDRESS)],
|
|
hostname="mock_hostname",
|
|
name="mock_name",
|
|
port=None,
|
|
properties={zeroconf.ATTR_PROPERTIES_ID: "aa:bb:cc:dd:ee:ff"},
|
|
type="mock_type",
|
|
),
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["errors"] is None
|
|
|
|
real_is_matching = YeelightConfigFlow.is_matching
|
|
return_values = []
|
|
|
|
def is_matching(self, other_flow) -> bool:
|
|
return_values.append(real_is_matching(self, other_flow))
|
|
return return_values[-1]
|
|
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
patch.object(
|
|
YeelightConfigFlow, "is_matching", wraps=is_matching, autospec=True
|
|
),
|
|
):
|
|
result2 = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": config_entries.SOURCE_DHCP},
|
|
data=dhcp.DhcpServiceInfo(
|
|
ip=IP_ADDRESS, macaddress="aabbccddeeff", hostname="mock_hostname"
|
|
),
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result2["type"] is FlowResultType.ABORT
|
|
assert result2["reason"] == "already_in_progress"
|
|
# Ensure the is_matching method returned True
|
|
assert return_values == [True]
|
|
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result3 = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": config_entries.SOURCE_DHCP},
|
|
data=dhcp.DhcpServiceInfo(
|
|
ip=IP_ADDRESS, macaddress="000000000000", hostname="mock_hostname"
|
|
),
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result3["type"] is FlowResultType.ABORT
|
|
assert result3["reason"] == "already_in_progress"
|
|
|
|
with (
|
|
_patch_discovery(no_device=True),
|
|
_patch_discovery_timeout(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", side_effect=CannotConnect),
|
|
):
|
|
result3 = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": config_entries.SOURCE_DHCP},
|
|
data=dhcp.DhcpServiceInfo(
|
|
ip="1.2.3.5", macaddress="000000000001", hostname="mock_hostname"
|
|
),
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result3["type"] is FlowResultType.ABORT
|
|
assert result3["reason"] == "cannot_connect"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("source", "data"),
|
|
[
|
|
(
|
|
config_entries.SOURCE_DHCP,
|
|
dhcp.DhcpServiceInfo(
|
|
ip=IP_ADDRESS, macaddress="aabbccddeeff", hostname="mock_hostname"
|
|
),
|
|
),
|
|
(
|
|
config_entries.SOURCE_HOMEKIT,
|
|
zeroconf.ZeroconfServiceInfo(
|
|
ip_address=ip_address(IP_ADDRESS),
|
|
ip_addresses=[ip_address(IP_ADDRESS)],
|
|
hostname="mock_hostname",
|
|
name="mock_name",
|
|
port=None,
|
|
properties={zeroconf.ATTR_PROPERTIES_ID: "aa:bb:cc:dd:ee:ff"},
|
|
type="mock_type",
|
|
),
|
|
),
|
|
],
|
|
)
|
|
async def test_discovered_by_dhcp_or_homekit(hass: HomeAssistant, source, data) -> None:
|
|
"""Test we can setup when discovered from dhcp or homekit."""
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": source}, data=data
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["errors"] is None
|
|
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE}.async_setup", return_value=True) as mock_async_setup,
|
|
patch(
|
|
f"{MODULE}.async_setup_entry", return_value=True
|
|
) as mock_async_setup_entry,
|
|
):
|
|
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
|
await hass.async_block_till_done()
|
|
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["data"] == {
|
|
CONF_HOST: IP_ADDRESS,
|
|
CONF_ID: "0x000000000015243f",
|
|
CONF_MODEL: MODEL,
|
|
}
|
|
assert mock_async_setup.called
|
|
assert mock_async_setup_entry.called
|
|
|
|
with (
|
|
_patch_discovery(no_device=True),
|
|
_patch_discovery_timeout(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", side_effect=CannotConnect),
|
|
):
|
|
result3 = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": source}, data=data
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result3["type"] is FlowResultType.ABORT
|
|
assert result3["reason"] == "already_configured"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("source", "data"),
|
|
[
|
|
(
|
|
config_entries.SOURCE_DHCP,
|
|
dhcp.DhcpServiceInfo(
|
|
ip=IP_ADDRESS, macaddress="aabbccddeeff", hostname="mock_hostname"
|
|
),
|
|
),
|
|
(
|
|
config_entries.SOURCE_HOMEKIT,
|
|
zeroconf.ZeroconfServiceInfo(
|
|
ip_address=ip_address(IP_ADDRESS),
|
|
ip_addresses=[ip_address(IP_ADDRESS)],
|
|
hostname="mock_hostname",
|
|
name="mock_name",
|
|
port=None,
|
|
properties={zeroconf.ATTR_PROPERTIES_ID: "aa:bb:cc:dd:ee:ff"},
|
|
type="mock_type",
|
|
),
|
|
),
|
|
],
|
|
)
|
|
async def test_discovered_by_dhcp_or_homekit_failed_to_get_id(
|
|
hass: HomeAssistant, source, data
|
|
) -> None:
|
|
"""Test we abort if we cannot get the unique id when discovered from dhcp or homekit."""
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(no_device=True),
|
|
_patch_discovery_timeout(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": source}, data=data
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "cannot_connect"
|
|
|
|
|
|
async def test_discovered_ssdp(hass: HomeAssistant) -> None:
|
|
"""Test we can setup when discovered from ssdp."""
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=SSDP_INFO
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["errors"] is None
|
|
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE}.async_setup", return_value=True) as mock_async_setup,
|
|
patch(
|
|
f"{MODULE}.async_setup_entry", return_value=True
|
|
) as mock_async_setup_entry,
|
|
):
|
|
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
|
await hass.async_block_till_done()
|
|
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["data"] == {
|
|
CONF_HOST: IP_ADDRESS,
|
|
CONF_ID: "0x000000000015243f",
|
|
CONF_MODEL: MODEL,
|
|
}
|
|
assert mock_async_setup.called
|
|
assert mock_async_setup_entry.called
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=SSDP_INFO
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
|
|
|
|
async def test_discovered_zeroconf(hass: HomeAssistant) -> None:
|
|
"""Test we can setup when discovered from zeroconf."""
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": config_entries.SOURCE_ZEROCONF},
|
|
data=ZEROCONF_DATA,
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["errors"] is None
|
|
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE}.async_setup", return_value=True) as mock_async_setup,
|
|
patch(
|
|
f"{MODULE}.async_setup_entry", return_value=True
|
|
) as mock_async_setup_entry,
|
|
):
|
|
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
|
await hass.async_block_till_done()
|
|
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["data"] == {
|
|
CONF_HOST: IP_ADDRESS,
|
|
CONF_ID: "0x000000000015243f",
|
|
CONF_MODEL: MODEL,
|
|
}
|
|
assert mock_async_setup.called
|
|
assert mock_async_setup_entry.called
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": config_entries.SOURCE_ZEROCONF},
|
|
data=ZEROCONF_DATA,
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": config_entries.SOURCE_SSDP},
|
|
data=SSDP_INFO,
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
|
|
|
|
async def test_discovery_updates_ip(hass: HomeAssistant) -> None:
|
|
"""Test discovery updates ip."""
|
|
config_entry = MockConfigEntry(
|
|
domain=DOMAIN, data={CONF_HOST: "1.2.2.3"}, unique_id=ID
|
|
)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": config_entries.SOURCE_ZEROCONF},
|
|
data=ZEROCONF_DATA,
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
assert config_entry.data[CONF_HOST] == IP_ADDRESS
|
|
|
|
|
|
async def test_discovery_updates_ip_no_reload_setup_in_progress(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test discovery updates ip does not reload if setup is an an error state."""
|
|
config_entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
data={CONF_HOST: "1.2.2.3"},
|
|
unique_id=ID,
|
|
state=config_entries.ConfigEntryState.SETUP_ERROR,
|
|
)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
patch(f"{MODULE}.async_setup_entry", return_value=True) as mock_setup_entry,
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": config_entries.SOURCE_ZEROCONF},
|
|
data=ZEROCONF_DATA,
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
assert config_entry.data[CONF_HOST] == IP_ADDRESS
|
|
assert len(mock_setup_entry.mock_calls) == 0
|
|
|
|
|
|
async def test_discovery_adds_missing_ip_id_only(hass: HomeAssistant) -> None:
|
|
"""Test discovery adds missing ip."""
|
|
config_entry = MockConfigEntry(domain=DOMAIN, data={CONF_ID: ID})
|
|
config_entry.add_to_hass(hass)
|
|
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": config_entries.SOURCE_ZEROCONF},
|
|
data=ZEROCONF_DATA,
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
assert config_entry.data[CONF_HOST] == IP_ADDRESS
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("source", "data"),
|
|
[
|
|
(
|
|
config_entries.SOURCE_DHCP,
|
|
dhcp.DhcpServiceInfo(
|
|
ip=IP_ADDRESS, macaddress="aabbccddeeff", hostname="mock_hostname"
|
|
),
|
|
),
|
|
(
|
|
config_entries.SOURCE_HOMEKIT,
|
|
zeroconf.ZeroconfServiceInfo(
|
|
ip_address=ip_address(IP_ADDRESS),
|
|
ip_addresses=[ip_address(IP_ADDRESS)],
|
|
hostname="mock_hostname",
|
|
name="mock_name",
|
|
port=None,
|
|
properties={zeroconf.ATTR_PROPERTIES_ID: "aa:bb:cc:dd:ee:ff"},
|
|
type="mock_type",
|
|
),
|
|
),
|
|
],
|
|
)
|
|
async def test_discovered_during_onboarding(hass: HomeAssistant, source, data) -> None:
|
|
"""Test we create a config entry when discovered during onboarding."""
|
|
mocked_bulb = _mocked_bulb()
|
|
with (
|
|
_patch_discovery(),
|
|
_patch_discovery_interval(),
|
|
patch(f"{MODULE_CONFIG_FLOW}.AsyncBulb", return_value=mocked_bulb),
|
|
patch(f"{MODULE}.async_setup", return_value=True) as mock_async_setup,
|
|
patch(
|
|
f"{MODULE}.async_setup_entry", return_value=True
|
|
) as mock_async_setup_entry,
|
|
patch(
|
|
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
|
|
) as mock_is_onboarded,
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": source}, data=data
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"] == {
|
|
CONF_HOST: IP_ADDRESS,
|
|
CONF_ID: "0x000000000015243f",
|
|
CONF_MODEL: MODEL,
|
|
}
|
|
assert mock_async_setup.called
|
|
assert mock_async_setup_entry.called
|
|
assert mock_is_onboarded.called
|