Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add zeroconf to TechnoVE integration (#108340) 

* Add zeroconf to TechnoVE integration

* Update homeassistant/components/technove/config_flow.py

Co-authored-by: Teemu R. <tpr@iki.fi>

* Update zeroconf test to test if update is called.

When a station is already configured and it is re-discovered through zeroconf, make sure we don't call its API for nothing.
pull/108704/head
Christophe Gagnier 2024-01-23 00:32:42 -05:00 committed by GitHub
parent 6fb86f179a
commit 4358c24edd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 236 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
homeassistant/components/technove/config_flow.py
View File

@ -5,8 +5,9 @@ from typing import Any
from technove import Station as TechnoVEStation, TechnoVE, TechnoVEConnectionError
import voluptuous as vol
from homeassistant.components import onboarding, zeroconf
from homeassistant.config_entries import ConfigFlow
from homeassistant.const import CONF_HOST
from homeassistant.const import CONF_HOST, CONF_MAC
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@ -16,6 +17,10 @@ from .const import DOMAIN
class TechnoVEConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for TechnoVE."""
VERSION = 1
discovered_host: str
discovered_station: TechnoVEStation
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
@ -44,6 +49,50 @@ class TechnoVEConfigFlow(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_zeroconf(
self, discovery_info: zeroconf.ZeroconfServiceInfo
) -> FlowResult:
"""Handle zeroconf discovery."""
# Abort quick if the device with provided mac is already configured
if mac := discovery_info.properties.get(CONF_MAC):
await self.async_set_unique_id(mac)
self._abort_if_unique_id_configured(
updates={CONF_HOST: discovery_info.host}
)
self.discovered_host = discovery_info.host
try:
self.discovered_station = await self._async_get_station(discovery_info.host)
except TechnoVEConnectionError:
return self.async_abort(reason="cannot_connect")
await self.async_set_unique_id(self.discovered_station.info.mac_address)
self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host})
self.context.update(
{
"title_placeholders": {"name": self.discovered_station.info.name},
}
)
return await self.async_step_zeroconf_confirm()
async def async_step_zeroconf_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by zeroconf."""
if user_input is not None or not onboarding.async_is_onboarded(self.hass):
return self.async_create_entry(
title=self.discovered_station.info.name,
data={
CONF_HOST: self.discovered_host,
},
)
return self.async_show_form(
step_id="zeroconf_confirm",
description_placeholders={"name": self.discovered_station.info.name},
)
async def _async_get_station(self, host: str) -> TechnoVEStation:
"""Get information from a TechnoVE station."""
api = TechnoVE(host, session=async_get_clientsession(self.hass))

3
homeassistant/components/technove/manifest.json
View File

@ -6,5 +6,6 @@
"documentation": "https://www.home-assistant.io/integrations/technove",
"integration_type": "device",
"iot_class": "local_polling",
"requirements": ["python-technove==1.1.1"]
"requirements": ["python-technove==1.1.1"],
"zeroconf": ["_technove-stations._tcp.local."]
}

4
homeassistant/components/technove/strings.json
View File

@ -10,6 +10,10 @@
"data_description": {
"host": "Hostname or IP address of your TechnoVE station."
}
},
"zeroconf_confirm": {
"description": "Do you want to add the TechnoVE Station named `{name}` to Home Assistant?",
"title": "Discovered TechnoVE station"
}
},
"error": {

5
homeassistant/generated/zeroconf.py
View File

@ -705,6 +705,11 @@ ZEROCONF = {
"domain": "system_bridge",
},
],
"_technove-stations._tcp.local.": [
{
"domain": "technove",
},
],
"_touch-able._tcp.local.": [
{
"domain": "apple_tv",

10
tests/components/technove/conftest.py
View File

@ -31,6 +31,16 @@ def mock_setup_entry() -> Generator[AsyncMock, None, None]:
yield mock_setup
@pytest.fixture
def mock_onboarding() -> Generator[MagicMock, None, None]:
"""Mock that Home Assistant is currently onboarding."""
with patch(
"homeassistant.components.onboarding.async_is_onboarded",
return_value=False,
) as mock_onboarding:
yield mock_onboarding
@pytest.fixture
def device_fixture() -> TechnoVEStation:
"""Return the device fixture for a specific device."""

168
tests/components/technove/test_config_flow.py
View File

@ -1,13 +1,15 @@
"""Tests for the TechnoVE config flow."""
from unittest.mock import MagicMock
from ipaddress import ip_address
from unittest.mock import AsyncMock, MagicMock
import pytest
from technove import TechnoVEConnectionError
from homeassistant.components import zeroconf
from homeassistant.components.technove.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST
from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@ -102,3 +104,163 @@ async def test_full_user_flow_with_error(
assert result["data"][CONF_HOST] == "192.168.1.123"
assert "result" in result
assert result["result"].unique_id == "AA:AA:AA:AA:AA:BB"
@pytest.mark.usefixtures("mock_setup_entry", "mock_technove")
async def test_full_zeroconf_flow_implementation(hass: HomeAssistant) -> None:
"""Test the full manual user flow from start to finish."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.123"),
ip_addresses=[ip_address("192.168.1.123")],
hostname="example.local.",
name="mock_name",
port=None,
properties={CONF_MAC: "AA:AA:AA:AA:AA:BB"},
type="mock_type",
),
)
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
assert result.get("description_placeholders") == {CONF_NAME: "TechnoVE Station"}
assert result.get("step_id") == "zeroconf_confirm"
assert result.get("type") == FlowResultType.FORM
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result2.get("title") == "TechnoVE Station"
assert result2.get("type") == FlowResultType.CREATE_ENTRY
assert "data" in result2
assert result2["data"][CONF_HOST] == "192.168.1.123"
assert "result" in result2
assert result2["result"].unique_id == "AA:AA:AA:AA:AA:BB"
@pytest.mark.usefixtures("mock_technove")
async def test_zeroconf_during_onboarding(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_onboarding: MagicMock,
) -> None:
"""Test we create a config entry when discovered during onboarding."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.123"),
ip_addresses=[ip_address("192.168.1.123")],
hostname="example.local.",
name="mock_name",
port=None,
properties={CONF_MAC: "AA:AA:AA:AA:AA:BB"},
type="mock_type",
),
)
assert result.get("title") == "TechnoVE Station"
assert result.get("type") == FlowResultType.CREATE_ENTRY
assert result.get("data") == {CONF_HOST: "192.168.1.123"}
assert "result" in result
assert result["result"].unique_id == "AA:AA:AA:AA:AA:BB"
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_onboarding.mock_calls) == 1
async def test_zeroconf_connection_error(
hass: HomeAssistant, mock_technove: MagicMock
) -> None:
"""Test we abort zeroconf flow on TechnoVE connection error."""
mock_technove.update.side_effect = TechnoVEConnectionError
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.123"),
ip_addresses=[ip_address("192.168.1.123")],
hostname="example.local.",
name="mock_name",
port=None,
properties={CONF_MAC: "AA:AA:AA:AA:AA:BB"},
type="mock_type",
),
)
assert result.get("type") == FlowResultType.ABORT
assert result.get("reason") == "cannot_connect"
@pytest.mark.usefixtures("mock_technove")
async def test_user_station_exists_abort(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test we abort zeroconf flow if TechnoVE station already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_HOST: "192.168.1.123"},
)
assert result.get("type") == FlowResultType.ABORT
assert result.get("reason") == "already_configured"
@pytest.mark.usefixtures("mock_technove")
async def test_zeroconf_without_mac_station_exists_abort(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test we abort zeroconf flow if TechnoVE station already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.123"),
ip_addresses=[ip_address("192.168.1.123")],
hostname="example.local.",
name="mock_name",
port=None,
properties={},
type="mock_type",
),
)
assert result.get("type") == FlowResultType.ABORT
assert result.get("reason") == "already_configured"
@pytest.mark.usefixtures("mock_technove")
async def test_zeroconf_with_mac_station_exists_abort(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_technove: MagicMock
) -> None:
"""Test we abort zeroconf flow if TechnoVE station already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=zeroconf.ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.123"),
ip_addresses=[ip_address("192.168.1.123")],
hostname="example.local.",
name="mock_name",
port=None,
properties={CONF_MAC: "AA:AA:AA:AA:AA:BB"},
type="mock_type",
),
)
mock_technove.update.assert_not_called()
assert result.get("type") == FlowResultType.ABORT
assert result.get("reason") == "already_configured"