jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-16 17:57:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve dhcp discovery for somfy_mylink to handle multiple devices (#45166)

Browse Source 
* Improve dhcp discovery for somfy_mylink to handle multiple devices

* Fix ip

* normalize

* logging debug

* force str

* we need a str

* fix tests

* add tests

* pylint
This commit is contained in:
J. Nick Koston 2021-01-15 08:47:25 -10:00 committed by GitHub
parent 7ce897f373
commit cffb1458a1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 144 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
homeassistant/components/somfy_mylink/config_flow.py
View File

@ -7,8 +7,10 @@ from somfy_mylink_synergy import SomfyMyLinkSynergy
import voluptuous as vol
from homeassistant import config_entries, core, exceptions
from homeassistant.components.dhcp import HOSTNAME, IP_ADDRESS, MAC_ADDRESS
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import callback
from homeassistant.helpers.device_registry import format_mac
from .const import (
CONF_REVERSE,
@ -23,19 +25,11 @@ from .const import DOMAIN # pylint:disable=unused-import
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_SYSTEM_ID): int,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): int,
}
)
async def validate_input(hass: core.HomeAssistant, data):
"""Validate the user input allows us to connect.
Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
Data has the keys from schema with values provided by the user.
"""
somfy_mylink = SomfyMyLinkSynergy(
data[CONF_SYSTEM_ID], data[CONF_HOST], data[CONF_PORT]
@ -47,6 +41,7 @@ async def validate_input(hass: core.HomeAssistant, data):
raise CannotConnect from ex
if not status_info or "error" in status_info:
_LOGGER.debug("Auth error: %s", status_info)
raise InvalidAuth
return {"title": f"MyLink {data[CONF_HOST]}"}
@ -58,32 +53,59 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_ASSUMED
def __init__(self):
"""Initialize the somfy_mylink flow."""
self.host = None
self.mac = None
self.ip_address = None
async def async_step_dhcp(self, dhcp_discovery):
"""Handle dhcp discovery."""
if self._host_already_configured(dhcp_discovery[IP_ADDRESS]):
return self.async_abort(reason="already_configured")
formatted_mac = format_mac(dhcp_discovery[MAC_ADDRESS])
await self.async_set_unique_id(format_mac(formatted_mac))
self._abort_if_unique_id_configured(
updates={CONF_HOST: dhcp_discovery[IP_ADDRESS]}
)
self.host = dhcp_discovery[HOSTNAME]
self.mac = formatted_mac
self.ip_address = dhcp_discovery[IP_ADDRESS]
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {"ip": self.ip_address, "mac": self.mac}
return await self.async_step_user()
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
errors = {}
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
if user_input is not None:
if self._host_already_configured(user_input[CONF_HOST]):
return self.async_abort(reason="already_configured")
if self._host_already_configured(user_input[CONF_HOST]):
return self.async_abort(reason="already_configured")
try:
info = await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return self.async_create_entry(title=info["title"], data=user_input)
try:
info = await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return self.async_create_entry(title=info["title"], data=user_input)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST, default=self.ip_address): str,
vol.Required(CONF_SYSTEM_ID): str,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): int,
}
),
errors=errors,
)
async def async_step_import(self, user_input):

1
homeassistant/components/somfy_mylink/strings.json
View File

@ -1,6 +1,7 @@
{
"title": "Somfy MyLink",
"config": {
"flow_title": "Somfy MyLink {mac} ({ip})",
"step": {
"user": {
"description": "The System ID can be obtained in the MyLink app under Integration by selecting any non-Cloud service.",

1
homeassistant/components/somfy_mylink/translations/en.json
View File

@ -1,6 +1,7 @@
{
"title": "Somfy MyLink",
"config": {
"flow_title": "Somfy MyLink {mac} ({ip})",
"step": {
"user": {
"description": "The System ID can be obtained in the MyLink app under Integration by selecting any non-Cloud service.",

102
tests/components/somfy_mylink/test_config_flow.py
View File

@ -5,6 +5,7 @@ from unittest.mock import patch
import pytest
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.dhcp import HOSTNAME, IP_ADDRESS, MAC_ADDRESS
from homeassistant.components.somfy_mylink.const import (
CONF_DEFAULT_REVERSE,
CONF_ENTITY_CONFIG,
@ -41,7 +42,7 @@ async def test_form_user(hass):
{
CONF_HOST: "1.1.1.1",
CONF_PORT: 1234,
CONF_SYSTEM_ID: 456,
CONF_SYSTEM_ID: "456",
},
)
await hass.async_block_till_done()
@ -51,7 +52,7 @@ async def test_form_user(hass):
assert result2["data"] == {
CONF_HOST: "1.1.1.1",
CONF_PORT: 1234,
CONF_SYSTEM_ID: 456,
CONF_SYSTEM_ID: "456",
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@ -86,7 +87,7 @@ async def test_form_user_already_configured(hass):
{
CONF_HOST: "1.1.1.1",
CONF_PORT: 1234,
CONF_SYSTEM_ID: 456,
CONF_SYSTEM_ID: "456",
},
)
await hass.async_block_till_done()
@ -195,7 +196,7 @@ async def test_form_import_already_exists(hass):
data={
CONF_HOST: "1.1.1.1",
CONF_PORT: 1234,
CONF_SYSTEM_ID: 456,
CONF_SYSTEM_ID: "456",
},
)
await hass.async_block_till_done()
@ -224,7 +225,7 @@ async def test_form_invalid_auth(hass):
{
CONF_HOST: "1.1.1.1",
CONF_PORT: 1234,
CONF_SYSTEM_ID: 456,
CONF_SYSTEM_ID: "456",
},
)
@ -247,7 +248,7 @@ async def test_form_cannot_connect(hass):
{
CONF_HOST: "1.1.1.1",
CONF_PORT: 1234,
CONF_SYSTEM_ID: 456,
CONF_SYSTEM_ID: "456",
},
)
@ -270,7 +271,7 @@ async def test_form_unknown_error(hass):
{
CONF_HOST: "1.1.1.1",
CONF_PORT: 1234,
CONF_SYSTEM_ID: 456,
CONF_SYSTEM_ID: "456",
},
)
@ -284,7 +285,7 @@ async def test_options_not_loaded(hass):
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "1.1.1.1", CONF_PORT: 12, CONF_SYSTEM_ID: 46},
data={CONF_HOST: "1.1.1.1", CONF_PORT: 12, CONF_SYSTEM_ID: "46"},
)
config_entry.add_to_hass(hass)
@ -304,7 +305,7 @@ async def test_options_with_targets(hass, reversed):
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "1.1.1.1", CONF_PORT: 12, CONF_SYSTEM_ID: 46},
data={CONF_HOST: "1.1.1.1", CONF_PORT: 12, CONF_SYSTEM_ID: "46"},
)
config_entry.add_to_hass(hass)
@ -363,7 +364,7 @@ async def test_form_import_with_entity_config_modify_options(hass, reversed):
data={
CONF_HOST: "1.1.1.1",
CONF_PORT: 1234,
CONF_SYSTEM_ID: 456,
CONF_SYSTEM_ID: "456",
CONF_DEFAULT_REVERSE: True,
CONF_ENTITY_CONFIG: {"cover.xyz": {CONF_REVERSE: False}},
},
@ -422,3 +423,84 @@ async def test_form_import_with_entity_config_modify_options(hass, reversed):
}
await hass.async_block_till_done()
async def test_form_user_already_configured_from_dhcp(hass):
"""Test we abort if already configured from dhcp."""
await setup.async_setup_component(hass, "persistent_notification", {})
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "1.1.1.1", CONF_PORT: 12, CONF_SYSTEM_ID: 46},
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.somfy_mylink.config_flow.SomfyMyLinkSynergy.status_info",
return_value={"any": "data"},
), patch(
"homeassistant.components.somfy_mylink.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.somfy_mylink.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data={
IP_ADDRESS: "1.1.1.1",
MAC_ADDRESS: "AA:BB:CC:DD:EE:FF",
HOSTNAME: "somfy_eeff",
},
)
await hass.async_block_till_done()
assert result["type"] == "abort"
assert len(mock_setup.mock_calls) == 0
assert len(mock_setup_entry.mock_calls) == 0
async def test_dhcp_discovery(hass):
"""Test we can process the discovery from dhcp."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data={
IP_ADDRESS: "1.1.1.1",
MAC_ADDRESS: "AA:BB:CC:DD:EE:FF",
HOSTNAME: "somfy_eeff",
},
)
assert result["type"] == "form"
assert result["errors"] == {}
with patch(
"homeassistant.components.somfy_mylink.config_flow.SomfyMyLinkSynergy.status_info",
return_value={"any": "data"},
), patch(
"homeassistant.components.somfy_mylink.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.somfy_mylink.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_HOST: "1.1.1.1",
CONF_PORT: 1234,
CONF_SYSTEM_ID: "456",
},
)
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["title"] == "MyLink 1.1.1.1"
assert result2["data"] == {
CONF_HOST: "1.1.1.1",
CONF_PORT: 1234,
CONF_SYSTEM_ID: "456",
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1