jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-08-05 11:38:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add zeroconf flow to Homee (#149820)

Browse Source 
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Markus Adrario 2025-08-04 12:26:22 +02:00 committed by GitHub
parent afffe0b08b
commit cbf4130bff
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 343 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

157
homeassistant/components/homee/config_flow.py
View File

@ -11,10 +11,16 @@ from pyHomee import (
)
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import SOURCE_USER, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from .const import DOMAIN
from .const import (
DOMAIN,
RESULT_CANNOT_CONNECT,
RESULT_INVALID_AUTH,
RESULT_UNKNOWN_ERROR,
)
_LOGGER = logging.getLogger(__name__)
@ -33,60 +39,137 @@ class HomeeConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1
homee: Homee
_host: str
_name: str
_reauth_host: str
_reauth_username: str
async def _connect_homee(self) -> dict[str, str]:
errors: dict[str, str] = {}
try:
await self.homee.get_access_token()
except HomeeConnectionFailedException:
errors["base"] = RESULT_CANNOT_CONNECT
except HomeeAuthenticationFailedException:
errors["base"] = RESULT_INVALID_AUTH
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = RESULT_UNKNOWN_ERROR
else:
_LOGGER.info("Got access token for homee")
self.hass.loop.create_task(self.homee.run())
_LOGGER.debug("Homee task created")
await self.homee.wait_until_connected()
_LOGGER.info("Homee connected")
self.homee.disconnect()
_LOGGER.debug("Homee disconnecting")
await self.homee.wait_until_disconnected()
_LOGGER.info("Homee config successfully tested")
await self.async_set_unique_id(
self.homee.settings.uid, raise_on_progress=self.source != SOURCE_USER
)
self._abort_if_unique_id_configured()
_LOGGER.info("Created new homee entry with ID %s", self.homee.settings.uid)
return errors
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial user step."""
errors: dict[str, str] = {}
errors = {}
if user_input is not None:
self.homee = Homee(
user_input[CONF_HOST],
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
)
errors = await self._connect_homee()
try:
await self.homee.get_access_token()
except HomeeConnectionFailedException:
errors["base"] = "cannot_connect"
except HomeeAuthenticationFailedException:
errors["base"] = "invalid_auth"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
_LOGGER.info("Got access token for homee")
self.hass.loop.create_task(self.homee.run())
_LOGGER.debug("Homee task created")
await self.homee.wait_until_connected()
_LOGGER.info("Homee connected")
self.homee.disconnect()
_LOGGER.debug("Homee disconnecting")
await self.homee.wait_until_disconnected()
_LOGGER.info("Homee config successfully tested")
await self.async_set_unique_id(self.homee.settings.uid)
self._abort_if_unique_id_configured()
_LOGGER.info(
"Created new homee entry with ID %s", self.homee.settings.uid
)
if not errors:
return self.async_create_entry(
title=f"{self.homee.settings.homee_name} ({self.homee.host})",
data=user_input,
)
return self.async_show_form(
step_id="user",
data_schema=AUTH_SCHEMA,
errors=errors,
)
async def async_step_zeroconf(
self, discovery_info: ZeroconfServiceInfo
) -> ConfigFlowResult:
"""Handle zeroconf discovery."""
# Ensure that an IPv4 address is received
self._host = discovery_info.host
self._name = discovery_info.hostname[6:18]
if discovery_info.ip_address.version == 6:
return self.async_abort(reason="ipv6_address")
await self.async_set_unique_id(self._name)
self._abort_if_unique_id_configured(updates={CONF_HOST: self._host})
# Cause an auth-error to see if homee is reachable.
self.homee = Homee(
self._host,
"dummy_username",
"dummy_password",
)
errors = await self._connect_homee()
if errors["base"] != RESULT_INVALID_AUTH:
return self.async_abort(reason=RESULT_CANNOT_CONNECT)
self.context["title_placeholders"] = {"name": self._name, "host": self._host}
return await self.async_step_zeroconf_confirm()
async def async_step_zeroconf_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm the configuration of the device."""
errors: dict[str, str] = {}
if user_input is not None:
self.homee = Homee(
self._host,
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
)
errors = await self._connect_homee()
if not errors:
return self.async_create_entry(
title=f"{self.homee.settings.homee_name} ({self.homee.host})",
data={
CONF_HOST: self._host,
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
},
)
return self.async_show_form(
step_id="zeroconf_confirm",
data_schema=vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
),
errors=errors,
description_placeholders={
CONF_HOST: self._name,
},
last_step=True,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
@ -108,12 +191,12 @@ class HomeeConfigFlow(ConfigFlow, domain=DOMAIN):
try:
await self.homee.get_access_token()
except HomeeConnectionFailedException:
errors["base"] = "cannot_connect"
errors["base"] = RESULT_CANNOT_CONNECT
except HomeeAuthenticationFailedException:
errors["base"] = "invalid_auth"
errors["base"] = RESULT_INVALID_AUTH
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
errors["base"] = RESULT_UNKNOWN_ERROR
else:
self.hass.loop.create_task(self.homee.run())
await self.homee.wait_until_connected()
@ -161,12 +244,12 @@ class HomeeConfigFlow(ConfigFlow, domain=DOMAIN):
try:
await self.homee.get_access_token()
except HomeeConnectionFailedException:
errors["base"] = "cannot_connect"
errors["base"] = RESULT_CANNOT_CONNECT
except HomeeAuthenticationFailedException:
errors["base"] = "invalid_auth"
errors["base"] = RESULT_INVALID_AUTH
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
errors["base"] = RESULT_UNKNOWN_ERROR
else:
self.hass.loop.create_task(self.homee.run())
await self.homee.wait_until_connected()

5
homeassistant/components/homee/const.py
View File

@ -20,6 +20,11 @@ from homeassistant.const import (
# General
DOMAIN = "homee"
# Error strings
RESULT_CANNOT_CONNECT = "cannot_connect"
RESULT_INVALID_AUTH = "invalid_auth"
RESULT_UNKNOWN_ERROR = "unknown"
# Sensor mappings
HOMEE_UNIT_TO_HA_UNIT = {
"": None,

8
homeassistant/components/homee/manifest.json
View File

@ -8,5 +8,11 @@
"iot_class": "local_push",
"loggers": ["homee"],
"quality_scale": "silver",
"requirements": ["pyHomee==1.2.10"]
"requirements": ["pyHomee==1.2.10"],
"zeroconf": [
{
"type": "_ssh._tcp.local.",
"name": "homee-*"
}
]
}

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

@ -46,6 +46,17 @@
"data_description": {
"host": "[%key:component::homee::config::step::user::data_description::host%]"
}
},
"zeroconf_confirm": {
"title": "Configure discovered homee {host}",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"username": "[%key:component::homee::config::step::user::data_description::username%]",
"password": "[%key:component::homee::config::step::user::data_description::password%]"
}
}
}
},

4
homeassistant/generated/zeroconf.py generated
View File

@ -890,6 +890,10 @@ ZEROCONF = {
},
],
"_ssh._tcp.local.": [
{
"domain": "homee",
"name": "homee-*",
},
{
"domain": "smappee",
"name": "smappee1*",

248
tests/components/homee/test_config_flow.py
View File

@ -1,15 +1,23 @@
"""Test the Homee config flow."""
from ipaddress import ip_address
from unittest.mock import AsyncMock
from pyHomee import HomeeAuthFailedException, HomeeConnectionFailedException
import pytest
from homeassistant.components.homee.const import DOMAIN
from homeassistant import config_entries
from homeassistant.components.homee.const import (
DOMAIN,
RESULT_CANNOT_CONNECT,
RESULT_INVALID_AUTH,
RESULT_UNKNOWN_ERROR,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from .conftest import (
HOMEE_ID,
@ -24,6 +32,24 @@ from .conftest import (
from tests.common import MockConfigEntry
PARAMETRIZED_ERRORS = (
("side_eff", "error"),
[
(
HomeeConnectionFailedException("connection timed out"),
{"base": RESULT_CANNOT_CONNECT},
),
(
HomeeAuthFailedException("wrong username or password"),
{"base": RESULT_INVALID_AUTH},
),
(
Exception,
{"base": RESULT_UNKNOWN_ERROR},
),
],
)
@pytest.mark.usefixtures("mock_homee", "mock_config_entry", "mock_setup_entry")
async def test_config_flow(
@ -58,23 +84,7 @@ async def test_config_flow(
assert result["result"].unique_id == HOMEE_ID
@pytest.mark.parametrize(
("side_eff", "error"),
[
(
HomeeConnectionFailedException("connection timed out"),
{"base": "cannot_connect"},
),
(
HomeeAuthFailedException("wrong username or password"),
{"base": "invalid_auth"},
),
(
Exception,
{"base": "unknown"},
),
],
)
@pytest.mark.parametrize(*PARAMETRIZED_ERRORS)
async def test_config_flow_errors(
hass: HomeAssistant,
mock_homee: AsyncMock,
@ -140,6 +150,172 @@ async def test_flow_already_configured(
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("mock_homee", "mock_config_entry")
async def test_zeroconf_success(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_homee: AsyncMock,
) -> None:
"""Test zeroconf discovery flow."""
mock_homee.get_access_token.side_effect = HomeeAuthFailedException(
"wrong username or password"
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=ZeroconfServiceInfo(
name=f"homee-{HOMEE_ID}._ssh._tcp.local.",
type="_ssh._tcp.local.",
hostname=f"homee-{HOMEE_ID}.local.",
ip_address=ip_address(HOMEE_IP),
ip_addresses=[ip_address(HOMEE_IP)],
port=22,
properties={},
),
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "zeroconf_confirm"
assert result["handler"] == DOMAIN
mock_setup_entry.assert_not_called()
mock_homee.get_access_token.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: TESTUSER,
CONF_PASSWORD: TESTPASS,
},
)
assert result["data"] == {
CONF_HOST: HOMEE_IP,
CONF_USERNAME: TESTUSER,
CONF_PASSWORD: TESTPASS,
}
mock_setup_entry.assert_called_once()
@pytest.mark.parametrize(*PARAMETRIZED_ERRORS)
async def test_zeroconf_confirm_errors(
hass: HomeAssistant,
mock_homee: AsyncMock,
side_eff: Exception,
error: dict[str, str],
) -> None:
"""Test zeroconf discovery flow errors."""
mock_homee.get_access_token.side_effect = HomeeAuthFailedException(
"wrong username or password"
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=ZeroconfServiceInfo(
name=f"homee-{HOMEE_ID}._ssh._tcp.local.",
type="_ssh._tcp.local.",
hostname=f"homee-{HOMEE_ID}.local.",
ip_address=ip_address(HOMEE_IP),
ip_addresses=[ip_address(HOMEE_IP)],
port=22,
properties={},
),
)
flow_id = result["flow_id"]
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "zeroconf_confirm"
assert result["handler"] == DOMAIN
mock_homee.get_access_token.side_effect = side_eff
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: TESTUSER,
CONF_PASSWORD: TESTPASS,
},
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == error
mock_homee.get_access_token.side_effect = None
result = await hass.config_entries.flow.async_configure(
flow_id,
user_input={
CONF_USERNAME: TESTUSER,
CONF_PASSWORD: TESTPASS,
},
)
assert result["type"] == FlowResultType.CREATE_ENTRY
async def test_zeroconf_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test zeroconf discovery flow when already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=ZeroconfServiceInfo(
name=f"homee-{HOMEE_ID}._ssh._tcp.local.",
type="_ssh._tcp.local.",
hostname=f"homee-{HOMEE_ID}.local.",
ip_address=ip_address(HOMEE_IP),
ip_addresses=[ip_address(HOMEE_IP)],
port=22,
properties={},
),
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("side_eff", "ip", "reason"),
[
(
HomeeConnectionFailedException("connection timed out"),
HOMEE_IP,
RESULT_CANNOT_CONNECT,
),
(Exception, HOMEE_IP, RESULT_CANNOT_CONNECT),
(None, "2001:db8::1", "ipv6_address"),
],
)
async def test_zeroconf_errors(
hass: HomeAssistant,
mock_homee: AsyncMock,
side_eff: Exception,
ip: str,
reason: str,
) -> None:
"""Test zeroconf discovery flow with an IPv6 address."""
mock_homee.get_access_token.side_effect = side_eff
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=ZeroconfServiceInfo(
name=f"homee-{HOMEE_ID}._ssh._tcp.local.",
type="_ssh._tcp.local.",
hostname=f"homee-{HOMEE_ID}.local.",
ip_address=ip_address(ip),
ip_addresses=[ip_address(ip)],
port=22,
properties={},
),
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == reason
@pytest.mark.usefixtures("mock_homee", "mock_setup_entry")
async def test_reauth_success(
hass: HomeAssistant,
@ -171,23 +347,7 @@ async def test_reauth_success(
assert mock_config_entry.data[CONF_PASSWORD] == NEW_TESTPASS
@pytest.mark.parametrize(
("side_eff", "error"),
[
(
HomeeConnectionFailedException("connection timed out"),
{"base": "cannot_connect"},
),
(
HomeeAuthFailedException("wrong username or password"),
{"base": "invalid_auth"},
),
(
Exception,
{"base": "unknown"},
),
],
)
@pytest.mark.parametrize(*PARAMETRIZED_ERRORS)
async def test_reauth_errors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
@ -296,23 +456,7 @@ async def test_reconfigure_success(
assert mock_config_entry.data[CONF_PASSWORD] == TESTPASS
@pytest.mark.parametrize(
("side_eff", "error"),
[
(
HomeeConnectionFailedException("connection timed out"),
{"base": "cannot_connect"},
),
(
HomeeAuthFailedException("wrong username or password"),
{"base": "invalid_auth"},
),
(
Exception,
{"base": "unknown"},
),
],
)
@pytest.mark.parametrize(*PARAMETRIZED_ERRORS)
async def test_reconfigure_errors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,