starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Improve Avea Bluetooth discovery flow (#172623)

Browse Source
This commit is contained in:
Sören
2026-05-30 09:49:36 -05:00
committed by GitHub
parent 740778f00b
commit 840243db9c
2 changed files with 82 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/avea/config_flow.py
+13 -6
View File
@@ -8,6 +8,7 @@ import avea
from bleak.exc import BleakError
import voluptuous as vol
from homeassistant.components import bluetooth
from homeassistant.components.bluetooth import (
BluetoothServiceInfoBleak,
async_discovered_service_info,
@@ -66,6 +67,15 @@ def _is_avea_discovery(discovery_info: BluetoothServiceInfoBleak) -> bool:
return AVEA_SERVICE_UUID in discovery_info.service_uuids
def _discovery_label(discovery_info: BluetoothServiceInfoBleak) -> str:
"""Return a label for a discovered Avea bulb."""
if (
name := _normalize_name(discovery_info.name)
) and name != discovery_info.address:
return f"{name} ({discovery_info.address})"
return discovery_info.address
class AveaConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Avea."""
@@ -150,6 +160,7 @@ class AveaConfigFlow(ConfigFlow, domain=DOMAIN):
if discovery := self._discovery_info:
self._discovered_devices[discovery.address] = discovery
else:
await bluetooth.async_request_active_scan(self.hass)
current_addresses = self._async_current_ids(include_ignore=False)
for discovery in async_discovered_service_info(self.hass):
if (
@@ -165,11 +176,10 @@ class AveaConfigFlow(ConfigFlow, domain=DOMAIN):
if self._discovery_info:
disc = self._discovery_info
label = f"{disc.name or disc.address} ({disc.address})"
data_schema = vol.Schema(
{
vol.Required(CONF_ADDRESS, default=disc.address): vol.In(
{disc.address: label}
{disc.address: _discovery_label(disc)}
)
}
)
@@ -178,10 +188,7 @@ class AveaConfigFlow(ConfigFlow, domain=DOMAIN):
{
vol.Required(CONF_ADDRESS): vol.In(
{
service_info.address: (
f"{service_info.name or service_info.address}"
f" ({service_info.address})"
)
service_info.address: _discovery_label(service_info)
for service_info in self._discovered_devices.values()
}
),
tests/components/avea/test_config_flow.py
+69 -8
View File
@@ -4,7 +4,7 @@ from unittest.mock import MagicMock, patch
import pytest
from homeassistant.components.avea.const import DOMAIN
from homeassistant.components.avea.const import AVEA_SERVICE_UUID, DOMAIN
from homeassistant.config_entries import SOURCE_BLUETOOTH, SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_ADDRESS, CONF_NAME
from homeassistant.core import HomeAssistant
@@ -12,7 +12,11 @@ from homeassistant.data_entry_flow import FlowResultType
from . import AVEA_DISCOVERY_INFO, NOT_AVEA_DISCOVERY_INFO
from tests.components.bluetooth import inject_bluetooth_service_info
from tests.components.bluetooth import (
generate_advertisement_data,
generate_ble_device,
inject_bluetooth_service_info,
)
pytestmark = pytest.mark.usefixtures("enable_bluetooth")
@@ -35,13 +39,22 @@ async def test_user_step_success(hass: HomeAssistant) -> None:
inject_bluetooth_service_info(hass, AVEA_DISCOVERY_INFO)
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with patch(
"homeassistant.components.avea.config_flow.bluetooth.async_request_active_scan"
) as mock_request_active_scan:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
assert result["data_schema"].schema[CONF_ADDRESS].container == {
AVEA_DISCOVERY_INFO.address: (
f"{AVEA_DISCOVERY_INFO.name} ({AVEA_DISCOVERY_INFO.address})"
)
}
mock_request_active_scan.assert_awaited_once_with(hass)
with (
patch(
@@ -67,14 +80,62 @@ async def test_user_step_no_devices_found(hass: HomeAssistant) -> None:
inject_bluetooth_service_info(hass, NOT_AVEA_DISCOVERY_INFO)
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with patch(
"homeassistant.components.avea.config_flow.bluetooth.async_request_active_scan"
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_devices_found"
async def test_user_step_unnamed_device_label(hass: HomeAssistant) -> None:
"""Test unnamed discovered devices are shown without duplicating the address."""
discovery_info = type(AVEA_DISCOVERY_INFO)(
name=AVEA_DISCOVERY_INFO.address,
address=AVEA_DISCOVERY_INFO.address,
rssi=-60,
manufacturer_data={},
service_uuids=[AVEA_SERVICE_UUID],
service_data={},
source="local",
device=generate_ble_device(
address=AVEA_DISCOVERY_INFO.address, name=AVEA_DISCOVERY_INFO.address
),
advertisement=generate_advertisement_data(
local_name=AVEA_DISCOVERY_INFO.address,
manufacturer_data={},
service_data={},
service_uuids=[AVEA_SERVICE_UUID],
),
time=0,
connectable=True,
tx_power=-127,
)
with (
patch(
"homeassistant.components.avea.config_flow.async_discovered_service_info",
return_value=[discovery_info],
),
patch(
"homeassistant.components.avea.config_flow.bluetooth.async_request_active_scan"
) as mock_request_active_scan,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["data_schema"].schema[CONF_ADDRESS].container == {
AVEA_DISCOVERY_INFO.address: AVEA_DISCOVERY_INFO.address
}
mock_request_active_scan.assert_awaited_once_with(hass)
async def test_user_step_cannot_connect_recovers(hass: HomeAssistant) -> None:
"""Test the user step recovers after a cannot connect error."""
inject_bluetooth_service_info(hass, AVEA_DISCOVERY_INFO)