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

Add LD2410 BLE integration ()

pull/85275/head
930913 2023-01-06 02:15:03 +00:00 committed by GitHub
parent 5b8b6167ac
commit 2507ec1f4b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 709 additions and 0 deletions

3
.coveragerc
View File

@ -670,6 +670,9 @@ omit =
homeassistant/components/lcn/helpers.py
homeassistant/components/lcn/scene.py
homeassistant/components/lcn/services.py
homeassistant/components/ld2410_ble/__init__.py
homeassistant/components/ld2410_ble/binary_sensor.py
homeassistant/components/ld2410_ble/coordinator.py
homeassistant/components/led_ble/__init__.py
homeassistant/components/led_ble/light.py
homeassistant/components/lg_netcast/media_player.py

1
.strict-typing
View File

@ -178,6 +178,7 @@ homeassistant.components.lacrosse_view.*
homeassistant.components.lametric.*
homeassistant.components.laundrify.*
homeassistant.components.lcn.*
homeassistant.components.ld2410_ble.*
homeassistant.components.lidarr.*
homeassistant.components.lifx.*
homeassistant.components.light.*

2
CODEOWNERS
View File

@ -627,6 +627,8 @@ build.json @home-assistant/supervisor
/tests/components/laundrify/ @xLarry
/homeassistant/components/lcn/ @alengwenus
/tests/components/lcn/ @alengwenus
/homeassistant/components/ld2410_ble/ @930913
/tests/components/ld2410_ble/ @930913
/homeassistant/components/led_ble/ @bdraco
/tests/components/led_ble/ @bdraco
/homeassistant/components/lg_netcast/ @Drafteed

94
homeassistant/components/ld2410_ble/__init__.py Normal file
View File

@ -0,0 +1,94 @@
"""The LD2410 BLE integration."""
import logging
from bleak_retry_connector import BleakError, get_device
from ld2410_ble import LD2410BLE
from homeassistant.components import bluetooth
from homeassistant.components.bluetooth.match import ADDRESS, BluetoothCallbackMatcher
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from .const import DOMAIN
from .coordinator import LD2410BLECoordinator
from .models import LD2410BLEData
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up LD2410 BLE from a config entry."""
address: str = entry.data[CONF_ADDRESS]
ble_device = bluetooth.async_ble_device_from_address(
hass, address.upper(), True
) or await get_device(address)
if not ble_device:
raise ConfigEntryNotReady(
f"Could not find LD2410B device with address {address}"
)
ld2410_ble = LD2410BLE(ble_device)
coordinator = LD2410BLECoordinator(hass, ld2410_ble)
try:
await ld2410_ble.initialise()
except BleakError as exc:
raise ConfigEntryNotReady(
f"Could not initialise LD2410B device with address {address}"
) from exc
@callback
def _async_update_ble(
service_info: bluetooth.BluetoothServiceInfoBleak,
change: bluetooth.BluetoothChange,
) -> None:
"""Update from a ble callback."""
ld2410_ble.set_ble_device_and_advertisement_data(
service_info.device, service_info.advertisement
)
entry.async_on_unload(
bluetooth.async_register_callback(
hass,
_async_update_ble,
BluetoothCallbackMatcher({ADDRESS: address}),
bluetooth.BluetoothScanningMode.ACTIVE,
)
)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = LD2410BLEData(
entry.title, ld2410_ble, coordinator
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
async def _async_stop(event: Event) -> None:
"""Close the connection."""
await ld2410_ble.stop()
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_stop)
)
return True
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
data: LD2410BLEData = hass.data[DOMAIN][entry.entry_id]
if entry.title != data.title:
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
data: LD2410BLEData = hass.data[DOMAIN].pop(entry.entry_id)
await data.device.stop()
return unload_ok

81
homeassistant/components/ld2410_ble/binary_sensor.py Normal file
View File

@ -0,0 +1,81 @@
"""LD2410 BLE integration binary sensor platform."""
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import LD2410BLE, LD2410BLECoordinator
from .const import DOMAIN
from .models import LD2410BLEData
ENTITY_DESCRIPTIONS = [
BinarySensorEntityDescription(
key="is_moving",
device_class=BinarySensorDeviceClass.MOTION,
has_entity_name=True,
name="Motion",
),
BinarySensorEntityDescription(
key="is_static",
device_class=BinarySensorDeviceClass.OCCUPANCY,
has_entity_name=True,
name="Occupancy",
),
]
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the platform for LD2410BLE."""
data: LD2410BLEData = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
LD2410BLEBinarySensor(data.coordinator, data.device, entry.title, description)
for description in ENTITY_DESCRIPTIONS
)
class LD2410BLEBinarySensor(CoordinatorEntity, BinarySensorEntity):
"""Moving/static sensor for LD2410BLE."""
def __init__(
self,
coordinator: LD2410BLECoordinator,
device: LD2410BLE,
name: str,
description: BinarySensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._coordinator = coordinator
self._key = description.key
self._device = device
self.entity_description = description
self._attr_unique_id = f"{device.address}_{self._key}"
self._attr_device_info = DeviceInfo(
name=name,
connections={(dr.CONNECTION_BLUETOOTH, device.address)},
)
self._attr_is_on = getattr(self._device, self._key)
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._attr_is_on = getattr(self._device, self._key)
self.async_write_ha_state()
@property
def available(self) -> bool:
"""Unavailable if coordinator isn't connected."""
return self._coordinator.connected and super().available

112
homeassistant/components/ld2410_ble/config_flow.py Normal file
View File

@ -0,0 +1,112 @@
"""Config flow for LD2410BLE integration."""
from __future__ import annotations
import logging
from typing import Any
from bluetooth_data_tools import human_readable_name
from ld2410_ble import BLEAK_EXCEPTIONS, LD2410BLE
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.bluetooth import (
BluetoothServiceInfoBleak,
async_discovered_service_info,
)
from homeassistant.const import CONF_ADDRESS
from homeassistant.data_entry_flow import FlowResult
from .const import DOMAIN, LOCAL_NAMES
_LOGGER = logging.getLogger(__name__)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for LD2410 BLE."""
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self._discovery_info: BluetoothServiceInfoBleak | None = None
self._discovered_devices: dict[str, BluetoothServiceInfoBleak] = {}
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfoBleak
) -> FlowResult:
"""Handle the bluetooth discovery step."""
await self.async_set_unique_id(discovery_info.address)
self._abort_if_unique_id_configured()
self._discovery_info = discovery_info
self.context["title_placeholders"] = {
"name": human_readable_name(
None, discovery_info.name, discovery_info.address
)
}
return await self.async_step_user()
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the user step to pick discovered device."""
errors: dict[str, str] = {}
if user_input is not None:
address = user_input[CONF_ADDRESS]
discovery_info = self._discovered_devices[address]
local_name = discovery_info.name
await self.async_set_unique_id(
discovery_info.address, raise_on_progress=False
)
self._abort_if_unique_id_configured()
ld2410_ble = LD2410BLE(discovery_info.device)
try:
await ld2410_ble.initialise()
except BLEAK_EXCEPTIONS:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected error")
errors["base"] = "unknown"
else:
await ld2410_ble.stop()
return self.async_create_entry(
title=local_name,
data={
CONF_ADDRESS: discovery_info.address,
},
)
if discovery := self._discovery_info:
self._discovered_devices[discovery.address] = discovery
else:
current_addresses = self._async_current_ids()
for discovery in async_discovered_service_info(self.hass):
if (
discovery.address in current_addresses
or discovery.address in self._discovered_devices
or not any(
discovery.name.startswith(local_name)
for local_name in LOCAL_NAMES
)
):
continue
self._discovered_devices[discovery.address] = discovery
if not self._discovered_devices:
return self.async_abort(reason="no_unconfigured_devices")
data_schema = vol.Schema(
{
vol.Required(CONF_ADDRESS): vol.In(
{
service_info.address: f"{service_info.name} ({service_info.address})"
for service_info in self._discovered_devices.values()
}
),
}
)
return self.async_show_form(
step_id="user",
data_schema=data_schema,
errors=errors,
)

5
homeassistant/components/ld2410_ble/const.py Normal file
View File

@ -0,0 +1,5 @@
"""Constants for the LD2410 BLE integration."""
DOMAIN = "ld2410_ble"
LOCAL_NAMES = {"HLK-LD2410B"}

40
homeassistant/components/ld2410_ble/coordinator.py Normal file
View File

@ -0,0 +1,40 @@
"""Data coordinator for receiving LD2410B updates."""
import logging
from ld2410_ble import LD2410BLE, LD2410BLEState
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class LD2410BLECoordinator(DataUpdateCoordinator):
"""Data coordinator for receiving LD2410B updates."""
def __init__(self, hass: HomeAssistant, ld2410_ble: LD2410BLE) -> None:
"""Initialise the coordinator."""
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
)
self._ld2410_ble = ld2410_ble
ld2410_ble.register_callback(self._async_handle_update)
ld2410_ble.register_disconnected_callback(self._async_handle_disconnect)
self.connected = False
@callback
def _async_handle_update(self, state: LD2410BLEState) -> None:
"""Just trigger the callbacks."""
self.connected = True
self.async_set_updated_data(True)
@callback
def _async_handle_disconnect(self) -> None:
"""Trigger the callbacks for disconnected."""
self.connected = False
self.async_update_listeners()

12
homeassistant/components/ld2410_ble/manifest.json Normal file
View File

@ -0,0 +1,12 @@
{
"domain": "ld2410_ble",
"name": "LD2410 BLE",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/ld2410_ble/",
"requirements": ["bluetooth-data-tools==0.3.0", "ld2410-ble==0.1.1"],
"dependencies": ["bluetooth"],
"codeowners": ["@930913"],
"bluetooth": [{ "local_name": "HLK-LD2410B_*" }],
"integration_type": "device",
"iot_class": "local_push"
}

17
homeassistant/components/ld2410_ble/models.py Normal file
View File

@ -0,0 +1,17 @@
"""The ld2410 ble integration models."""
from __future__ import annotations
from dataclasses import dataclass
from ld2410_ble import LD2410BLE
from .coordinator import LD2410BLECoordinator
@dataclass
class LD2410BLEData:
"""Data for the ld2410 ble integration."""
title: str
device: LD2410BLE
coordinator: LD2410BLECoordinator

23
homeassistant/components/ld2410_ble/strings.json Normal file
View File

@ -0,0 +1,23 @@
{
"config": {
"flow_title": "{name}",
"step": {
"user": {
"data": {
"address": "Bluetooth address"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"not_supported": "Device not supported",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"no_unconfigured_devices": "No unconfigured devices found.",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
}
}
}

23
homeassistant/components/ld2410_ble/translations/en.json Normal file
View File

@ -0,0 +1,23 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"already_in_progress": "Configuration flow is already in progress",
"no_devices_found": "No devices found on the network",
"no_unconfigured_devices": "No unconfigured devices found.",
"not_supported": "Device not supported"
},
"error": {
"cannot_connect": "Failed to connect",
"unknown": "Unexpected error"
},
"flow_title": "{name}",
"step": {
"user": {
"data": {
"address": "Bluetooth address"
}
}
}
}
}

4
homeassistant/generated/bluetooth.py
View File

@ -201,6 +201,10 @@ BLUETOOTH: list[dict[str, bool | str | int | list[int]]] = [
"domain": "keymitt_ble",
"local_name": "mib*",
},
{
"domain": "ld2410_ble",
"local_name": "HLK-LD2410B_*",
},
{
"domain": "led_ble",
"local_name": "LEDnet*",

1
homeassistant/generated/config_flows.py
View File

@ -221,6 +221,7 @@ FLOWS = {
"landisgyr_heat_meter",
"launch_library",
"laundrify",
"ld2410_ble",
"led_ble",
"lg_soundbar",
"lidarr",

6
homeassistant/generated/integrations.json
View File

@ -2775,6 +2775,12 @@
"config_flow": false,
"iot_class": "local_push"
},
"ld2410_ble": {
"name": "LD2410 BLE",
"integration_type": "device",
"config_flow": true,
"iot_class": "local_push"
},
"led_ble": {
"name": "LED BLE",
"integration_type": "hub",

10
mypy.ini
View File

@ -1534,6 +1534,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.ld2410_ble.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.lidarr.*]
check_untyped_defs = true
disallow_incomplete_defs = true

4
requirements_all.txt
View File

@ -462,6 +462,7 @@ bluetooth-adapters==0.15.2
bluetooth-auto-recovery==1.0.3
# homeassistant.components.bluetooth
# homeassistant.components.ld2410_ble
# homeassistant.components.led_ble
bluetooth-data-tools==0.3.1
@ -1019,6 +1020,9 @@ lakeside==0.12
# homeassistant.components.laundrify
laundrify_aio==1.1.2
# homeassistant.components.ld2410_ble
ld2410-ble==0.1.1
# homeassistant.components.led_ble
led-ble==1.0.0

4
requirements_test_all.txt
View File

@ -376,6 +376,7 @@ bluetooth-adapters==0.15.2
bluetooth-auto-recovery==1.0.3
# homeassistant.components.bluetooth
# homeassistant.components.ld2410_ble
# homeassistant.components.led_ble
bluetooth-data-tools==0.3.1
@ -760,6 +761,9 @@ lacrosse-view==0.0.9
# homeassistant.components.laundrify
laundrify_aio==1.1.2
# homeassistant.components.ld2410_ble
ld2410-ble==0.1.1
# homeassistant.components.led_ble
led-ble==1.0.0

37
tests/components/ld2410_ble/__init__.py Normal file
View File

@ -0,0 +1,37 @@
"""Tests for the LD2410 BLE Bluetooth integration."""
from bleak.backends.device import BLEDevice
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from tests.components.bluetooth import generate_advertisement_data
LD2410_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
name="HLK-LD2410B_EEFF",
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
manufacturer_data={},
service_uuids=[],
service_data={},
source="local",
device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="HLK-LD2410B_EEFF"),
advertisement=generate_advertisement_data(),
time=0,
connectable=True,
)
NOT_LD2410_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
name="Not",
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
manufacturer_data={
33: b"\x00\x00\xd1\xf0b;\xd8\x1dE\xd6\xba\xeeL\xdd]\xf5\xb2\xe9",
21: b"\x061\x00Z\x8f\x93\xb2\xec\x85\x06\x00i\x00\x02\x02Q\xed\x1d\xf0",
},
service_uuids=[],
service_data={},
source="local",
device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Aug"),
advertisement=generate_advertisement_data(),
time=0,
connectable=True,
)

8
tests/components/ld2410_ble/conftest.py Normal file
View File

@ -0,0 +1,8 @@
"""ld2410_ble session fixtures."""
import pytest
@pytest.fixture(autouse=True)
def mock_bluetooth(enable_bluetooth):
"""Auto mock bluetooth."""

222
tests/components/ld2410_ble/test_config_flow.py Normal file
View File

@ -0,0 +1,222 @@
"""Test the LD2410 BLE Bluetooth config flow."""
from unittest.mock import patch
from bleak import BleakError
from homeassistant import config_entries
from homeassistant.components.ld2410_ble.const import DOMAIN
from homeassistant.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import LD2410_BLE_DISCOVERY_INFO, NOT_LD2410_BLE_DISCOVERY_INFO
from tests.common import MockConfigEntry
async def test_user_step_success(hass: HomeAssistant) -> None:
"""Test user step success path."""
with patch(
"homeassistant.components.ld2410_ble.config_flow.async_discovered_service_info",
return_value=[NOT_LD2410_BLE_DISCOVERY_INFO, LD2410_BLE_DISCOVERY_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
with patch(
"homeassistant.components.ld2410_ble.config_flow.LD2410BLE.initialise",
), patch(
"homeassistant.components.ld2410_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == LD2410_BLE_DISCOVERY_INFO.name
assert result2["data"] == {
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
}
assert result2["result"].unique_id == LD2410_BLE_DISCOVERY_INFO.address
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_step_no_devices_found(hass: HomeAssistant) -> None:
"""Test user step with no devices found."""
with patch(
"homeassistant.components.ld2410_ble.config_flow.async_discovered_service_info",
return_value=[NOT_LD2410_BLE_DISCOVERY_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "no_unconfigured_devices"
async def test_user_step_no_new_devices_found(hass: HomeAssistant) -> None:
"""Test user step with only existing devices found."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
},
unique_id=LD2410_BLE_DISCOVERY_INFO.address,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.ld2410_ble.config_flow.async_discovered_service_info",
return_value=[LD2410_BLE_DISCOVERY_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "no_unconfigured_devices"
async def test_user_step_cannot_connect(hass: HomeAssistant) -> None:
"""Test user step and we cannot connect."""
with patch(
"homeassistant.components.ld2410_ble.config_flow.async_discovered_service_info",
return_value=[LD2410_BLE_DISCOVERY_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
with patch(
"homeassistant.components.ld2410_ble.config_flow.LD2410BLE.initialise",
side_effect=BleakError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "cannot_connect"}
with patch(
"homeassistant.components.ld2410_ble.config_flow.LD2410BLE.initialise",
), patch(
"homeassistant.components.ld2410_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
},
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == LD2410_BLE_DISCOVERY_INFO.name
assert result3["data"] == {
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
}
assert result3["result"].unique_id == LD2410_BLE_DISCOVERY_INFO.address
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_step_unknown_exception(hass: HomeAssistant) -> None:
"""Test user step with an unknown exception."""
with patch(
"homeassistant.components.ld2410_ble.config_flow.async_discovered_service_info",
return_value=[NOT_LD2410_BLE_DISCOVERY_INFO, LD2410_BLE_DISCOVERY_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
with patch(
"homeassistant.components.ld2410_ble.config_flow.LD2410BLE.initialise",
side_effect=RuntimeError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "unknown"}
with patch(
"homeassistant.components.ld2410_ble.config_flow.LD2410BLE.initialise",
), patch(
"homeassistant.components.ld2410_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
},
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == LD2410_BLE_DISCOVERY_INFO.name
assert result3["data"] == {
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
}
assert result3["result"].unique_id == LD2410_BLE_DISCOVERY_INFO.address
assert len(mock_setup_entry.mock_calls) == 1
async def test_bluetooth_step_success(hass: HomeAssistant) -> None:
"""Test bluetooth step success path."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=LD2410_BLE_DISCOVERY_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
with patch(
"homeassistant.components.ld2410_ble.config_flow.LD2410BLE.initialise",
), patch(
"homeassistant.components.ld2410_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == LD2410_BLE_DISCOVERY_INFO.name
assert result2["data"] == {
CONF_ADDRESS: LD2410_BLE_DISCOVERY_INFO.address,
}
assert result2["result"].unique_id == LD2410_BLE_DISCOVERY_INFO.address
assert len(mock_setup_entry.mock_calls) == 1