Add zeroconf detection to devolo Home Control (#47934)

Co-authored-by: Markus Bong <2Fake1987@gmail.com>
This commit is contained in:
Guido Schmitz 2021-04-22 20:38:56 +02:00 committed by GitHub
parent 34b258e812
commit 77372d9094
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 161 additions and 53 deletions

View file

@ -12,14 +12,20 @@ from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, EVENT_HOMEASSISTAN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .const import CONF_MYDEVOLO, DOMAIN, GATEWAY_SERIAL_PATTERN, PLATFORMS
from .const import (
CONF_MYDEVOLO,
DEFAULT_MYDEVOLO,
DOMAIN,
GATEWAY_SERIAL_PATTERN,
PLATFORMS,
)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up the devolo account from a config entry."""
hass.data.setdefault(DOMAIN, {})
mydevolo = _mydevolo(entry.data)
mydevolo = configure_mydevolo(entry.data)
credentials_valid = await hass.async_add_executor_job(mydevolo.credentials_valid)
@ -92,10 +98,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload
def _mydevolo(conf: dict) -> Mydevolo:
def configure_mydevolo(conf: dict) -> Mydevolo:
"""Configure mydevolo."""
mydevolo = Mydevolo()
mydevolo.user = conf[CONF_USERNAME]
mydevolo.password = conf[CONF_PASSWORD]
mydevolo.url = conf[CONF_MYDEVOLO]
mydevolo.url = conf.get(CONF_MYDEVOLO, DEFAULT_MYDEVOLO)
return mydevolo

View file

@ -1,14 +1,20 @@
"""Config flow to configure the devolo home control integration."""
import logging
from devolo_home_control_api.mydevolo import Mydevolo
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback
from homeassistant.helpers.typing import DiscoveryInfoType
from .const import CONF_MYDEVOLO, DEFAULT_MYDEVOLO, DOMAIN
from . import configure_mydevolo
from .const import ( # pylint:disable=unused-import
CONF_MYDEVOLO,
DEFAULT_MYDEVOLO,
DOMAIN,
SUPPORTED_MODEL_TYPES,
)
_LOGGER = logging.getLogger(__name__)
@ -29,22 +35,30 @@ class DevoloHomeControlFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_user(self, user_input=None):
"""Handle a flow initiated by the user."""
if self.show_advanced_options:
self.data_schema = {
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Required(CONF_MYDEVOLO, default=DEFAULT_MYDEVOLO): str,
}
self.data_schema[
vol.Required(CONF_MYDEVOLO, default=DEFAULT_MYDEVOLO)
] = str
if user_input is None:
return self._show_form(user_input)
user = user_input[CONF_USERNAME]
password = user_input[CONF_PASSWORD]
mydevolo = Mydevolo()
mydevolo.user = user
mydevolo.password = password
if self.show_advanced_options:
mydevolo.url = user_input[CONF_MYDEVOLO]
else:
mydevolo.url = DEFAULT_MYDEVOLO
return await self._connect_mydevolo(user_input)
async def async_step_zeroconf(self, discovery_info: DiscoveryInfoType):
"""Handle zeroconf discovery."""
# Check if it is a gateway
if discovery_info.get("properties", {}).get("MT") in SUPPORTED_MODEL_TYPES:
await self._async_handle_discovery_without_unique_id()
return await self.async_step_zeroconf_confirm()
return self.async_abort(reason="Not a devolo Home Control gateway.")
async def async_step_zeroconf_confirm(self, user_input=None):
"""Handle a flow initiated by zeroconf."""
if user_input is None:
return self._show_form(step_id="zeroconf_confirm")
return await self._connect_mydevolo(user_input)
async def _connect_mydevolo(self, user_input):
"""Connect to mydevolo."""
mydevolo = configure_mydevolo(conf=user_input)
credentials_valid = await self.hass.async_add_executor_job(
mydevolo.credentials_valid
)
@ -58,17 +72,17 @@ class DevoloHomeControlFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_create_entry(
title="devolo Home Control",
data={
CONF_PASSWORD: password,
CONF_USERNAME: user,
CONF_PASSWORD: mydevolo.password,
CONF_USERNAME: mydevolo.user,
CONF_MYDEVOLO: mydevolo.url,
},
)
@callback
def _show_form(self, errors=None):
def _show_form(self, errors=None, step_id="user"):
"""Show the form to the user."""
return self.async_show_form(
step_id="user",
step_id=step_id,
data_schema=vol.Schema(self.data_schema),
errors=errors if errors else {},
)

View file

@ -6,3 +6,4 @@ DEFAULT_MYDEVOLO = "https://www.mydevolo.com"
PLATFORMS = ["binary_sensor", "climate", "cover", "light", "sensor", "switch"]
CONF_MYDEVOLO = "mydevolo_url"
GATEWAY_SERIAL_PATTERN = re.compile(r"\d{16}")
SUPPORTED_MODEL_TYPES = ["2600", "2601"]

View file

@ -7,5 +7,6 @@
"config_flow": true,
"codeowners": ["@2Fake", "@Shutgun"],
"quality_scale": "silver",
"iot_class": "local_push"
"iot_class": "local_push",
"zeroconf": ["_dvl-deviceapi._tcp.local."]
}

View file

@ -11,10 +11,17 @@
"data": {
"username": "[%key:common::config_flow::data::email%] / devolo ID",
"password": "[%key:common::config_flow::data::password%]",
"mydevolo_url": "mydevolo [%key:common::config_flow::data::url%]",
"home_control_url": "Home Control [%key:common::config_flow::data::url%]"
"mydevolo_url": "mydevolo [%key:common::config_flow::data::url%]"
}
},
"zeroconf_confirm": {
"data": {
"username": "[%key:common::config_flow::data::email%] / devolo ID",
"password": "[%key:common::config_flow::data::password%]",
"mydevolo_url": "mydevolo [%key:common::config_flow::data::url%]"
}
}
}
}
}

View file

@ -9,7 +9,13 @@
"step": {
"user": {
"data": {
"home_control_url": "Home Control URL",
"mydevolo_url": "mydevolo URL",
"password": "Password",
"username": "Email / devolo ID"
}
},
"zeroconf_confirm": {
"data": {
"mydevolo_url": "mydevolo URL",
"password": "Password",
"username": "Email / devolo ID"

View file

@ -49,6 +49,11 @@ ZEROCONF = {
"domain": "daikin"
}
],
"_dvl-deviceapi._tcp.local.": [
{
"domain": "devolo_home_control"
}
],
"_elg._tcp.local.": [
{
"domain": "elgato"

View file

@ -0,0 +1,22 @@
"""Constants used for mocking data."""
DISCOVERY_INFO = {
"host": "192.168.0.1",
"port": 14791,
"hostname": "test.local.",
"type": "_dvl-deviceapi._tcp.local.",
"name": "dvl-deviceapi",
"properties": {
"Path": "/deviceapi",
"Version": "v0",
"Features": "",
"MT": "2600",
"SN": "1234567890",
"FirmwareVersion": "8.90.4",
"PlcMacAddress": "AA:BB:CC:DD:EE:FF",
},
}
DISCOVERY_INFO_WRONG_DEVOLO_DEVICE = {"properties": {"MT": "2700"}}
DISCOVERY_INFO_WRONG_DEVICE = {"properties": {"Features": ""}}

View file

@ -4,9 +4,15 @@ from unittest.mock import patch
import pytest
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.devolo_home_control.const import DOMAIN
from homeassistant.components.devolo_home_control.const import DEFAULT_MYDEVOLO, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from .const import (
DISCOVERY_INFO,
DISCOVERY_INFO_WRONG_DEVICE,
DISCOVERY_INFO_WRONG_DEVOLO_DEVICE,
)
from tests.common import MockConfigEntry
@ -19,28 +25,7 @@ async def test_form(hass):
assert result["type"] == "form"
assert result["errors"] == {}
with patch(
"homeassistant.components.devolo_home_control.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.devolo_home_control.config_flow.Mydevolo.uuid",
return_value="123456",
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"username": "test-username", "password": "test-password"},
)
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["title"] == "devolo Home Control"
assert result2["data"] == {
"username": "test-username",
"password": "test-password",
"mydevolo_url": "https://www.mydevolo.com",
}
assert len(mock_setup_entry.mock_calls) == 1
await _setup(hass, result)
@pytest.mark.credentials_invalid
@ -64,7 +49,7 @@ async def test_form_invalid_credentials(hass):
async def test_form_already_configured(hass):
"""Test if we get the error message on already configured."""
with patch(
"homeassistant.components.devolo_home_control.config_flow.Mydevolo.uuid",
"homeassistant.components.devolo_home_control.Mydevolo.uuid",
return_value="123456",
):
MockConfigEntry(domain=DOMAIN, unique_id="123456", data={}).add_to_hass(hass)
@ -89,7 +74,7 @@ async def test_form_advanced_options(hass):
"homeassistant.components.devolo_home_control.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.devolo_home_control.config_flow.Mydevolo.uuid",
"homeassistant.components.devolo_home_control.Mydevolo.uuid",
return_value="123456",
):
result2 = await hass.config_entries.flow.async_configure(
@ -111,3 +96,64 @@ async def test_form_advanced_options(hass):
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_show_zeroconf_form(hass):
"""Test that the zeroconf confirmation form is served."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=DISCOVERY_INFO,
)
assert result["step_id"] == "zeroconf_confirm"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
await _setup(hass, result)
async def test_zeroconf_wrong_device(hass):
"""Test that the zeroconf ignores wrong devices."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=DISCOVERY_INFO_WRONG_DEVOLO_DEVICE,
)
assert result["reason"] == "Not a devolo Home Control gateway."
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=DISCOVERY_INFO_WRONG_DEVICE,
)
assert result["reason"] == "Not a devolo Home Control gateway."
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
async def _setup(hass, result):
"""Finish configuration steps."""
with patch(
"homeassistant.components.devolo_home_control.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.devolo_home_control.Mydevolo.uuid",
return_value="123456",
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"username": "test-username", "password": "test-password"},
)
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["title"] == "devolo Home Control"
assert result2["data"] == {
"username": "test-username",
"password": "test-password",
"mydevolo_url": DEFAULT_MYDEVOLO,
}
assert len(mock_setup_entry.mock_calls) == 1