github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-09-06 10:29:55 +02:00
Code Releases Activity

Add DHCP discovery to Obihai (#88984)

Browse Source 
* Add DHCP discovery to Obihai

* Unique ID is MAC

* Move try blocks, cleanup

* Migrate existing unique_ids

* Use PyObihai to update Unique ID

* Auth then use get_device_mac

* Config flow changes

* Reworking flow according to feedback

* Cleanup
This commit is contained in:
Emory Penney 2023-04-03 12:17:56 -07:00 committed by GitHub
parent fa332668d6
commit 7c6a32ebb5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 312 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
homeassistant/components/obihai/__init__.py
View File

@ -1,9 +1,11 @@
"""The Obihai integration."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from .const import PLATFORMS
from .connectivity import ObihaiConnection
from .const import LOGGER, PLATFORMS
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@ -13,6 +15,32 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Migrate old entry."""
version = entry.version
LOGGER.debug("Migrating from version %s", version)
if version != 2:
requester = ObihaiConnection(
entry.data[CONF_HOST],
username=entry.data[CONF_USERNAME],
password=entry.data[CONF_PASSWORD],
)
await hass.async_add_executor_job(requester.update)
new_unique_id = await hass.async_add_executor_job(
requester.pyobihai.get_device_mac
)
hass.config_entries.async_update_entry(entry, unique_id=new_unique_id)
entry.version = 2
LOGGER.info("Migration to version %s successful", entry.version)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

122
homeassistant/components/obihai/config_flow.py
View File

@ -1,10 +1,14 @@
"""Config flow to configure the Obihai integration."""
from __future__ import annotations
from socket import gaierror, gethostbyname
from typing import Any
from pyobihai import PyObihai
import voluptuous as vol
from homeassistant.components import dhcp
from homeassistant.config_entries import ConfigFlow
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
@ -16,11 +20,11 @@ from .const import DEFAULT_PASSWORD, DEFAULT_USERNAME, DOMAIN
DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Optional(
vol.Required(
CONF_USERNAME,
default=DEFAULT_USERNAME,
): str,
vol.Optional(
vol.Required(
CONF_PASSWORD,
default=DEFAULT_PASSWORD,
): str,
@ -28,48 +32,122 @@ DATA_SCHEMA = vol.Schema(
)
async def async_validate_creds(hass: HomeAssistant, user_input: dict[str, Any]) -> bool:
async def async_validate_creds(
hass: HomeAssistant, user_input: dict[str, Any]
) -> PyObihai | None:
"""Manage Obihai options."""
return await hass.async_add_executor_job(
validate_auth,
user_input[CONF_HOST],
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
)
if user_input[CONF_USERNAME] and user_input[CONF_PASSWORD]:
return await hass.async_add_executor_job(
validate_auth,
user_input[CONF_HOST],
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
)
# Don't bother authenticating if we've already determined the credentials are invalid
return None
class ObihaiFlowHandler(ConfigFlow, domain=DOMAIN):
"""Config flow for Obihai."""
VERSION = 1
VERSION = 2
discovery_schema: vol.Schema | None = None
_dhcp_discovery_info: dhcp.DhcpServiceInfo | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
errors: dict[str, str] = {}
ip: str | None = None
if user_input is not None:
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
if await async_validate_creds(self.hass, user_input):
return self.async_create_entry(
title=user_input[CONF_HOST],
data=user_input,
)
errors["base"] = "cannot_connect"
try:
ip = gethostbyname(user_input[CONF_HOST])
except gaierror:
errors["base"] = "cannot_connect"
data_schema = self.add_suggested_values_to_schema(DATA_SCHEMA, user_input)
if ip:
if pyobihai := await async_validate_creds(self.hass, user_input):
device_mac = await self.hass.async_add_executor_job(
pyobihai.get_device_mac
)
await self.async_set_unique_id(device_mac)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input[CONF_HOST],
data=user_input,
)
errors["base"] = "invalid_auth"
data_schema = self.discovery_schema or DATA_SCHEMA
return self.async_show_form(
step_id="user",
errors=errors,
data_schema=data_schema,
data_schema=self.add_suggested_values_to_schema(data_schema, user_input),
)
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
"""Prepare configuration for a DHCP discovered Obihai."""
self._dhcp_discovery_info = discovery_info
return await self.async_step_dhcp_confirm()
async def async_step_dhcp_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Attempt to confirm."""
assert self._dhcp_discovery_info
await self.async_set_unique_id(self._dhcp_discovery_info.macaddress)
self._abort_if_unique_id_configured()
if user_input is None:
credentials = {
CONF_HOST: self._dhcp_discovery_info.ip,
CONF_PASSWORD: DEFAULT_PASSWORD,
CONF_USERNAME: DEFAULT_USERNAME,
}
if await async_validate_creds(self.hass, credentials):
self.discovery_schema = self.add_suggested_values_to_schema(
DATA_SCHEMA, credentials
)
else:
self.discovery_schema = self.add_suggested_values_to_schema(
DATA_SCHEMA,
{
CONF_HOST: self._dhcp_discovery_info.ip,
CONF_USERNAME: "",
CONF_PASSWORD: "",
},
)
# Show the confirmation dialog
return self.async_show_form(
step_id="dhcp_confirm",
data_schema=self.discovery_schema,
description_placeholders={CONF_HOST: self._dhcp_discovery_info.ip},
)
return await self.async_step_user(user_input=user_input)
# DEPRECATED
async def async_step_import(self, config: dict[str, Any]) -> FlowResult:
"""Handle a flow initialized by importing a config."""
self._async_abort_entries_match({CONF_HOST: config[CONF_HOST]})
if await async_validate_creds(self.hass, config):
try:
_ = gethostbyname(config[CONF_HOST])
except gaierror:
return self.async_abort(reason="cannot_connect")
if pyobihai := await async_validate_creds(self.hass, config):
device_mac = await self.hass.async_add_executor_job(pyobihai.get_device_mac)
await self.async_set_unique_id(device_mac)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=config.get(CONF_NAME, config[CONF_HOST]),
data={
@ -79,4 +157,4 @@ class ObihaiFlowHandler(ConfigFlow, domain=DOMAIN):
},
)
return self.async_abort(reason="cannot_connect")
return self.async_abort(reason="invalid_auth")

10
homeassistant/components/obihai/connectivity.py
View File

@ -1,4 +1,5 @@
"""Support for Obihai Connectivity."""
from __future__ import annotations
from pyobihai import PyObihai
@ -12,6 +13,7 @@ def get_pyobihai(
password: str,
) -> PyObihai:
"""Retrieve an authenticated PyObihai."""
return PyObihai(host, username, password)
@ -19,16 +21,17 @@ def validate_auth(
host: str,
username: str,
password: str,
) -> bool:
) -> PyObihai | None:
"""Test if the given setting works as expected."""
obi = get_pyobihai(host, username, password)
login = obi.check_account()
if not login:
LOGGER.debug("Invalid credentials")
return False
return None
return True
return obi
class ObihaiConnection:
@ -53,6 +56,7 @@ class ObihaiConnection:
def update(self) -> bool:
"""Validate connection and retrieve a list of sensors."""
if not self.pyobihai:
self.pyobihai = get_pyobihai(self.host, self.username, self.password)

5
homeassistant/components/obihai/manifest.json
View File

@ -3,6 +3,11 @@
"name": "Obihai",
"codeowners": ["@dshokouhi", "@ejpenney"],
"config_flow": true,
"dhcp": [
{
"macaddress": "9CADEF*"
}
],
"documentation": "https://www.home-assistant.io/integrations/obihai",
"iot_class": "local_polling",
"loggers": ["pyobihai"],

11
homeassistant/components/obihai/strings.json
View File

@ -7,10 +7,19 @@
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
}
},
"dhcp_confirm": {
"description": "Do you want to set up {host}?",
"data": {
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"

4
homeassistant/generated/dhcp.py
View File

@ -330,6 +330,10 @@ DHCP: list[dict[str, str | bool]] = [
"domain": "nuki",
"hostname": "nuki_bridge_*",
},
{
"domain": "obihai",
"macaddress": "9CADEF*",
},
{
"domain": "oncue",
"hostname": "kohlergen*",

26
tests/components/obihai/__init__.py
View File

@ -1,6 +1,6 @@
"""Tests for the Obihai Integration."""
from homeassistant.components import dhcp
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
USER_INPUT = {
@ -8,3 +8,27 @@ USER_INPUT = {
CONF_PASSWORD: "admin",
CONF_USERNAME: "admin",
}
DHCP_SERVICE_INFO = dhcp.DhcpServiceInfo(
hostname="obi200",
ip="192.168.1.100",
macaddress="9CADEF000000",
)
class MockPyObihai:
"""Mock PyObihai: Returns simulated PyObihai data."""
def get_device_mac(self):
"""Mock PyObihai.get_device_mac, return simulated MAC address."""
return DHCP_SERVICE_INFO.macaddress
def get_schema_suggestion(schema, key):
"""Get suggested value for key in voluptuous schema."""
for k in schema:
if k == key:
if k.description is None or "suggested_value" not in k.description:
return None
return k.description["suggested_value"]

13
tests/components/obihai/conftest.py
View File

@ -1,6 +1,7 @@
"""Define test fixtures for Obihai."""
from collections.abc import Generator
from socket import gaierror
from unittest.mock import AsyncMock, patch
import pytest
@ -9,7 +10,19 @@ import pytest
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.obihai.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_gaierror() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.obihai.config_flow.gethostbyname",
side_effect=gaierror(),
) as mock_setup_entry:
yield mock_setup_entry

124
tests/components/obihai/test_config_flow.py
View File

@ -1,14 +1,17 @@
"""Test the Obihai config flow."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant import config_entries
from homeassistant.components.obihai.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import USER_INPUT
from . import DHCP_SERVICE_INFO, USER_INPUT, MockPyObihai, get_schema_suggestion
VALIDATE_AUTH_PATCH = "homeassistant.components.obihai.config_flow.validate_auth"
@ -25,7 +28,7 @@ async def test_user_form(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> No
assert result["step_id"] == "user"
assert result["errors"] == {}
with patch("pyobihai.PyObihai.check_account"):
with patch(VALIDATE_AUTH_PATCH, return_value=MockPyObihai()):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
@ -40,7 +43,7 @@ async def test_user_form(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> No
async def test_auth_failure(hass: HomeAssistant) -> None:
"""Test we get the authentication error."""
"""Test we get the authentication error for user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@ -52,6 +55,24 @@ async def test_auth_failure(hass: HomeAssistant) -> None:
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "invalid_auth"
async def test_connect_failure(hass: HomeAssistant, mock_gaierror: Generator) -> None:
"""Test we get the connection error for user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "cannot_connect"
@ -59,7 +80,8 @@ async def test_auth_failure(hass: HomeAssistant) -> None:
async def test_yaml_import(hass: HomeAssistant) -> None:
"""Test we get the YAML imported."""
with patch(VALIDATE_AUTH_PATCH, return_value=True):
with patch(VALIDATE_AUTH_PATCH, return_value=MockPyObihai()):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
@ -71,8 +93,9 @@ async def test_yaml_import(hass: HomeAssistant) -> None:
assert "errors" not in result
async def test_yaml_import_fail(hass: HomeAssistant) -> None:
async def test_yaml_import_auth_fail(hass: HomeAssistant) -> None:
"""Test the YAML import fails."""
with patch(VALIDATE_AUTH_PATCH, return_value=False):
result = await hass.config_entries.flow.async_init(
DOMAIN,
@ -81,6 +104,97 @@ async def test_yaml_import_fail(hass: HomeAssistant) -> None:
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "invalid_auth"
assert "errors" not in result
async def test_yaml_import_connect_fail(
hass: HomeAssistant, mock_gaierror: Generator
) -> None:
"""Test the YAML import fails with invalid host."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=USER_INPUT,
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
assert "errors" not in result
async def test_dhcp_flow(hass: HomeAssistant) -> None:
"""Test that DHCP discovery works."""
with patch(
VALIDATE_AUTH_PATCH,
return_value=MockPyObihai(),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=DHCP_SERVICE_INFO,
context={"source": config_entries.SOURCE_DHCP},
)
flows = hass.config_entries.flow.async_progress()
assert result["type"] == FlowResultType.FORM
assert len(flows) == 1
assert (
get_schema_suggestion(result["data_schema"].schema, CONF_USERNAME)
== USER_INPUT[CONF_USERNAME]
)
assert (
get_schema_suggestion(result["data_schema"].schema, CONF_PASSWORD)
== USER_INPUT[CONF_PASSWORD]
)
assert (
get_schema_suggestion(result["data_schema"].schema, CONF_HOST)
== DHCP_SERVICE_INFO.ip
)
assert flows[0].get("context", {}).get("source") == config_entries.SOURCE_DHCP
# Verify we get dropped into the normal user flow with non-default credentials
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=USER_INPUT
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
async def test_dhcp_flow_auth_failure(hass: HomeAssistant) -> None:
"""Test that DHCP fails if creds aren't default."""
with patch(
VALIDATE_AUTH_PATCH,
return_value=False,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=DHCP_SERVICE_INFO,
context={"source": config_entries.SOURCE_DHCP},
)
assert result["step_id"] == "dhcp_confirm"
assert get_schema_suggestion(result["data_schema"].schema, CONF_USERNAME) == ""
assert get_schema_suggestion(result["data_schema"].schema, CONF_PASSWORD) == ""
assert (
get_schema_suggestion(result["data_schema"].schema, CONF_HOST)
== DHCP_SERVICE_INFO.ip
)
# Verify we get dropped into the normal user flow with non-default credentials
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: DHCP_SERVICE_INFO.ip,
CONF_USERNAME: "",
CONF_PASSWORD: "",
},
)
assert result["errors"]["base"] == "invalid_auth"
assert result["step_id"] == "user"