evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Yale Access Bluetooth integration (#76182)

Browse source
This commit is contained in:
J. Nick Koston 2022-08-10 10:08:02 -10:00 • committed by GitHub
parent 2f99d6a32d
commit bb0038319d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 1483 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.coveragerc
View file

@ -1483,6 +1483,10 @@ omit =
homeassistant/components/xiaomi_tv/media_player.py
homeassistant/components/xmpp/notify.py
homeassistant/components/xs1/*
homeassistant/components/yalexs_ble/__init__.py
homeassistant/components/yalexs_ble/entity.py
homeassistant/components/yalexs_ble/lock.py
homeassistant/components/yalexs_ble/util.py
homeassistant/components/yale_smart_alarm/__init__.py
homeassistant/components/yale_smart_alarm/alarm_control_panel.py
homeassistant/components/yale_smart_alarm/binary_sensor.py

2
CODEOWNERS
View file

@ -1246,6 +1246,8 @@ build.json @home-assistant/supervisor
/homeassistant/components/xmpp/ @fabaff @flowolf
/homeassistant/components/yale_smart_alarm/ @gjohansson-ST
/tests/components/yale_smart_alarm/ @gjohansson-ST
/homeassistant/components/yalexs_ble/ @bdraco
/tests/components/yalexs_ble/ @bdraco
/homeassistant/components/yamaha_musiccast/ @vigonotion @micha91
/tests/components/yamaha_musiccast/ @vigonotion @micha91
/homeassistant/components/yandex_transport/ @rishatik92 @devbis

91
homeassistant/components/yalexs_ble/__init__.py Normal file
View file

@ -0,0 +1,91 @@
"""The Yale Access Bluetooth integration."""
from __future__ import annotations
import asyncio
import async_timeout
from yalexs_ble import PushLock, local_name_is_unique
from homeassistant.components import bluetooth
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from .const import CONF_KEY, CONF_LOCAL_NAME, CONF_SLOT, DEVICE_TIMEOUT, DOMAIN
from .models import YaleXSBLEData
from .util import async_find_existing_service_info, bluetooth_callback_matcher
PLATFORMS: list[Platform] = [Platform.LOCK]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Yale Access Bluetooth from a config entry."""
local_name = entry.data[CONF_LOCAL_NAME]
address = entry.data[CONF_ADDRESS]
key = entry.data[CONF_KEY]
slot = entry.data[CONF_SLOT]
has_unique_local_name = local_name_is_unique(local_name)
push_lock = PushLock(local_name, address, None, key, slot)
id_ = local_name if has_unique_local_name else address
push_lock.set_name(f"{entry.title} ({id_})")
startup_event = asyncio.Event()
@callback
def _async_update_ble(
service_info: bluetooth.BluetoothServiceInfoBleak,
change: bluetooth.BluetoothChange,
) -> None:
"""Update from a ble callback."""
push_lock.update_advertisement(service_info.device, service_info.advertisement)
cancel_first_update = push_lock.register_callback(lambda *_: startup_event.set())
entry.async_on_unload(await push_lock.start())
# We may already have the advertisement, so check for it.
if service_info := async_find_existing_service_info(hass, local_name, address):
push_lock.update_advertisement(service_info.device, service_info.advertisement)
entry.async_on_unload(
bluetooth.async_register_callback(
hass,
_async_update_ble,
bluetooth_callback_matcher(local_name, push_lock.address),
bluetooth.BluetoothScanningMode.PASSIVE,
)
)
try:
async with async_timeout.timeout(DEVICE_TIMEOUT):
await startup_event.wait()
except asyncio.TimeoutError as ex:
raise ConfigEntryNotReady(
f"{push_lock.last_error}; "
f"Try moving the Bluetooth adapter closer to {local_name}"
) from ex
finally:
cancel_first_update()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = YaleXSBLEData(
entry.title, push_lock
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
return True
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
data: YaleXSBLEData = 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):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

245
homeassistant/components/yalexs_ble/config_flow.py Normal file
View file

@ -0,0 +1,245 @@
"""Config flow for Yale Access Bluetooth integration."""
from __future__ import annotations
import asyncio
import logging
from typing import Any
from bleak_retry_connector import BleakError, BLEDevice
import voluptuous as vol
from yalexs_ble import (
AuthError,
DisconnectedError,
PushLock,
ValidatedLockConfig,
local_name_is_unique,
)
from yalexs_ble.const import YALE_MFR_ID
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 AbortFlow, FlowResult
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.typing import DiscoveryInfoType
from homeassistant.loader import async_get_integration
from .const import CONF_KEY, CONF_LOCAL_NAME, CONF_SLOT, DOMAIN
from .util import async_get_service_info, human_readable_name
_LOGGER = logging.getLogger(__name__)
async def validate_lock(
local_name: str, device: BLEDevice, key: str, slot: int
) -> None:
"""Validate a lock."""
if len(key) != 32:
raise InvalidKeyFormat
try:
bytes.fromhex(key)
except ValueError as ex:
raise InvalidKeyFormat from ex
if not isinstance(slot, int) or slot < 0 or slot > 255:
raise InvalidKeyIndex
await PushLock(local_name, device.address, device, key, slot).validate()
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Yale Access Bluetooth."""
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self._discovery_info: BluetoothServiceInfoBleak | None = None
self._discovered_devices: dict[str, BluetoothServiceInfoBleak] = {}
self._lock_cfg: ValidatedLockConfig | None = None
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.context["local_name"] = discovery_info.name
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_integration_discovery(
self, discovery_info: DiscoveryInfoType
) -> FlowResult:
"""Handle a discovered integration."""
lock_cfg = ValidatedLockConfig(
discovery_info["name"],
discovery_info["address"],
discovery_info["serial"],
discovery_info["key"],
discovery_info["slot"],
)
# We do not want to raise on progress as integration_discovery takes
# precedence over other discovery flows since we already have the keys.
await self.async_set_unique_id(lock_cfg.address, raise_on_progress=False)
new_data = {CONF_KEY: lock_cfg.key, CONF_SLOT: lock_cfg.slot}
self._abort_if_unique_id_configured(updates=new_data)
for entry in self._async_current_entries():
if entry.data.get(CONF_LOCAL_NAME) == lock_cfg.local_name:
if self.hass.config_entries.async_update_entry(
entry, data={**entry.data, **new_data}
):
self.hass.async_create_task(
self.hass.config_entries.async_reload(entry.entry_id)
)
raise AbortFlow(reason="already_configured")
try:
self._discovery_info = await async_get_service_info(
self.hass, lock_cfg.local_name, lock_cfg.address
)
except asyncio.TimeoutError:
return self.async_abort(reason="no_devices_found")
for progress in self._async_in_progress(include_uninitialized=True):
# Integration discovery should abort other discovery types
# since it already has the keys and slots, and the other
# discovery types do not.
context = progress["context"]
if (
not context.get("active")
and context.get("local_name") == lock_cfg.local_name
or context.get("unique_id") == lock_cfg.address
):
self.hass.config_entries.flow.async_abort(progress["flow_id"])
self._lock_cfg = lock_cfg
self.context["title_placeholders"] = {
"name": human_readable_name(
lock_cfg.name, lock_cfg.local_name, self._discovery_info.address
)
}
return await self.async_step_integration_discovery_confirm()
async def async_step_integration_discovery_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a confirmation of discovered integration."""
assert self._discovery_info is not None
assert self._lock_cfg is not None
if user_input is not None:
return self.async_create_entry(
title=self._lock_cfg.name,
data={
CONF_LOCAL_NAME: self._discovery_info.name,
CONF_ADDRESS: self._discovery_info.address,
CONF_KEY: self._lock_cfg.key,
CONF_SLOT: self._lock_cfg.slot,
},
)
self._set_confirm_only()
return self.async_show_form(
step_id="integration_discovery_confirm",
description_placeholders={
"name": self._lock_cfg.name,
"address": self._discovery_info.address,
},
)
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:
self.context["active"] = True
address = user_input[CONF_ADDRESS]
discovery_info = self._discovered_devices[address]
local_name = discovery_info.name
key = user_input[CONF_KEY]
slot = user_input[CONF_SLOT]
await self.async_set_unique_id(
discovery_info.address, raise_on_progress=False
)
self._abort_if_unique_id_configured()
try:
await validate_lock(local_name, discovery_info.device, key, slot)
except InvalidKeyFormat:
errors[CONF_KEY] = "invalid_key_format"
except InvalidKeyIndex:
errors[CONF_SLOT] = "invalid_key_index"
except (DisconnectedError, AuthError, ValueError):
errors[CONF_KEY] = "invalid_auth"
except BleakError:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected error")
errors["base"] = "unknown"
else:
return self.async_create_entry(
title=local_name,
data={
CONF_LOCAL_NAME: discovery_info.name,
CONF_ADDRESS: discovery_info.address,
CONF_KEY: key,
CONF_SLOT: slot,
},
)
if discovery := self._discovery_info:
self._discovered_devices[discovery.address] = discovery
else:
current_addresses = self._async_current_ids()
current_unique_names = {
entry.data.get(CONF_LOCAL_NAME)
for entry in self._async_current_entries()
if local_name_is_unique(entry.data.get(CONF_LOCAL_NAME))
}
for discovery in async_discovered_service_info(self.hass):
if (
discovery.address in current_addresses
or discovery.name in current_unique_names
or discovery.address in self._discovered_devices
or YALE_MFR_ID not in discovery.manufacturer_data
):
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()
}
),
vol.Required(CONF_KEY): str,
vol.Required(CONF_SLOT): int,
}
)
integration = await async_get_integration(self.hass, DOMAIN)
return self.async_show_form(
step_id="user",
data_schema=data_schema,
errors=errors,
description_placeholders={"docs_url": integration.documentation},
)
class InvalidKeyFormat(HomeAssistantError):
"""Invalid key format."""
class InvalidKeyIndex(HomeAssistantError):
"""Invalid key index."""

9
homeassistant/components/yalexs_ble/const.py Normal file
View file

@ -0,0 +1,9 @@
"""Constants for the Yale Access Bluetooth integration."""
DOMAIN = "yalexs_ble"
CONF_LOCAL_NAME = "local_name"
CONF_KEY = "key"
CONF_SLOT = "slot"
DEVICE_TIMEOUT = 55

76
homeassistant/components/yalexs_ble/entity.py Normal file
View file

@ -0,0 +1,76 @@
"""The yalexs_ble integration entities."""
from __future__ import annotations
from yalexs_ble import ConnectionInfo, LockInfo, LockState
from homeassistant.components import bluetooth
from homeassistant.core import callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity import DeviceInfo, Entity
from .const import DOMAIN
from .models import YaleXSBLEData
class YALEXSBLEEntity(Entity):
"""Base class for yale xs ble entities."""
_attr_should_poll = False
def __init__(self, data: YaleXSBLEData) -> None:
"""Initialize the entity."""
self._data = data
self._device = device = data.lock
self._attr_available = False
lock_state = device.lock_state
lock_info = device.lock_info
connection_info = device.connection_info
assert lock_state is not None
assert connection_info is not None
assert lock_info is not None
self._attr_unique_id = device.address
self._attr_device_info = DeviceInfo(
name=data.title,
manufacturer=lock_info.manufacturer,
model=lock_info.model,
connections={(dr.CONNECTION_BLUETOOTH, device.address)},
identifiers={(DOMAIN, lock_info.serial)},
sw_version=lock_info.firmware,
)
if device.lock_state:
self._async_update_state(lock_state, lock_info, connection_info)
@callback
def _async_update_state(
self, new_state: LockState, lock_info: LockInfo, connection_info: ConnectionInfo
) -> None:
"""Update the state."""
self._attr_available = True
@callback
def _async_state_changed(
self, new_state: LockState, lock_info: LockInfo, connection_info: ConnectionInfo
) -> None:
"""Handle state changed."""
self._async_update_state(new_state, lock_info, connection_info)
self.async_write_ha_state()
@callback
def _async_device_unavailable(self, _address: str) -> None:
"""Handle device not longer being seen by the bluetooth stack."""
self._attr_available = False
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
self.async_on_remove(
bluetooth.async_track_unavailable(
self.hass, self._async_device_unavailable, self._device.address
)
)
self.async_on_remove(self._device.register_callback(self._async_state_changed))
return await super().async_added_to_hass()
async def async_update(self) -> None:
"""Request a manual update."""
await self._device.update()

62
homeassistant/components/yalexs_ble/lock.py Normal file
View file

@ -0,0 +1,62 @@
"""Support for Yale Access Bluetooth locks."""
from __future__ import annotations
from typing import Any
from yalexs_ble import ConnectionInfo, LockInfo, LockState, LockStatus
from homeassistant.components.lock import LockEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .entity import YALEXSBLEEntity
from .models import YaleXSBLEData
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up locks."""
data: YaleXSBLEData = hass.data[DOMAIN][entry.entry_id]
async_add_entities([YaleXSBLELock(data)])
class YaleXSBLELock(YALEXSBLEEntity, LockEntity):
"""A yale xs ble lock."""
_attr_has_entity_name = True
@callback
def _async_update_state(
self, new_state: LockState, lock_info: LockInfo, connection_info: ConnectionInfo
) -> None:
"""Update the state."""
self._attr_is_locked = False
self._attr_is_locking = False
self._attr_is_unlocking = False
self._attr_is_jammed = False
lock_state = new_state.lock
if lock_state == LockStatus.LOCKED:
self._attr_is_locked = True
elif lock_state == LockStatus.LOCKING:
self._attr_is_locking = True
elif lock_state == LockStatus.UNLOCKING:
self._attr_is_unlocking = True
elif lock_state in (
LockStatus.UNKNOWN_01,
LockStatus.UNKNOWN_06,
):
self._attr_is_jammed = True
super()._async_update_state(new_state, lock_info, connection_info)
async def async_unlock(self, **kwargs: Any) -> None:
"""Unlock the lock."""
return await self._device.unlock()
async def async_lock(self, **kwargs: Any) -> None:
"""Lock the lock."""
return await self._device.lock()

11
homeassistant/components/yalexs_ble/manifest.json Normal file
View file

@ -0,0 +1,11 @@
{
"domain": "yalexs_ble",
"name": "Yale Access Bluetooth",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/yalexs_ble",
"requirements": ["yalexs-ble==1.1.2"],
"dependencies": ["bluetooth"],
"codeowners": ["@bdraco"],
"bluetooth": [{ "manufacturer_id": 465 }],
"iot_class": "local_push"
}

14
homeassistant/components/yalexs_ble/models.py Normal file
View file

@ -0,0 +1,14 @@
"""The yalexs_ble integration models."""
from __future__ import annotations
from dataclasses import dataclass
from yalexs_ble import PushLock
@dataclass
class YaleXSBLEData:
"""Data for the yale xs ble integration."""
title: str
lock: PushLock

30
homeassistant/components/yalexs_ble/strings.json Normal file
View file

@ -0,0 +1,30 @@
{
"config": {
"flow_title": "{name}",
"step": {
"user": {
"description": "Check the documentation at {docs_url} for how to find the offline key.",
"data": {
"address": "Bluetooth address",
"key": "Offline Key (32-byte hex string)",
"slot": "Offline Key Slot (Integer between 0 and 255)"
}
},
"integration_discovery_confirm": {
"description": "Do you want to setup {name} over Bluetooth with address {address}?"
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"invalid_key_format": "The offline key must be a 32-byte hex string.",
"invalid_key_index": "The offline key slot must be an integer between 0 and 255."
},
"abort": {
"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%]"
}
}
}

30
homeassistant/components/yalexs_ble/translations/en.json Normal file
View file

@ -0,0 +1,30 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"no_devices_found": "No devices found on the network",
"no_unconfigured_devices": "No unconfigured devices found."
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"invalid_key_format": "The offline key must be a 32-byte hex string.",
"invalid_key_index": "The offline key slot must be an integer between 0 and 255.",
"unknown": "Unexpected error"
},
"flow_title": "{name}",
"step": {
"integration_discovery_confirm": {
"description": "Do you want to setup {name} over Bluetooth with address {address}?"
},
"user": {
"data": {
"address": "Bluetooth address",
"key": "Offline Key (32-byte hex string)",
"slot": "Offline Key Slot (Integer between 0 and 255)"
},
"description": "Check the documentation at {docs_url} for how to find the offline key."
}
}
}
}

69
homeassistant/components/yalexs_ble/util.py Normal file
View file

@ -0,0 +1,69 @@
"""The yalexs_ble integration models."""
from __future__ import annotations
from yalexs_ble import local_name_is_unique
from homeassistant.components.bluetooth import (
BluetoothScanningMode,
BluetoothServiceInfoBleak,
async_discovered_service_info,
async_process_advertisements,
)
from homeassistant.components.bluetooth.match import (
ADDRESS,
LOCAL_NAME,
BluetoothCallbackMatcher,
)
from homeassistant.core import HomeAssistant, callback
from .const import DEVICE_TIMEOUT
def bluetooth_callback_matcher(
local_name: str, address: str
) -> BluetoothCallbackMatcher:
"""Return a BluetoothCallbackMatcher for the given local_name and address."""
if local_name_is_unique(local_name):
return BluetoothCallbackMatcher({LOCAL_NAME: local_name})
return BluetoothCallbackMatcher({ADDRESS: address})
@callback
def async_find_existing_service_info(
hass: HomeAssistant, local_name: str, address: str
) -> BluetoothServiceInfoBleak | None:
"""Return the service info for the given local_name and address."""
has_unique_local_name = local_name_is_unique(local_name)
for service_info in async_discovered_service_info(hass):
device = service_info.device
if (
has_unique_local_name and device.name == local_name
) or device.address == address:
return service_info
return None
async def async_get_service_info(
hass: HomeAssistant, local_name: str, address: str
) -> BluetoothServiceInfoBleak:
"""Wait for the service info for the given local_name and address."""
if service_info := async_find_existing_service_info(hass, local_name, address):
return service_info
return await async_process_advertisements(
hass,
lambda service_info: True,
bluetooth_callback_matcher(local_name, address),
BluetoothScanningMode.ACTIVE,
DEVICE_TIMEOUT,
)
def short_address(address: str) -> str:
"""Convert a Bluetooth address to a short address."""
split_address = address.replace("-", ":").split(":")
return f"{split_address[-2].upper()}{split_address[-1].upper()}"[-4:]
def human_readable_name(name: str | None, local_name: str, address: str) -> str:
"""Return a human readable name for the given name, local_name, and address."""
return f"{name or local_name} ({short_address(address)})"

4
homeassistant/generated/bluetooth.py
View file

@ -107,5 +107,9 @@ BLUETOOTH: list[dict[str, str | int | list[int]]] = [
{
"domain": "xiaomi_ble",
"service_data_uuid": "0000fe95-0000-1000-8000-00805f9b34fb"
},
{
"domain": "yalexs_ble",
"manufacturer_id": 465
}
]

1
homeassistant/generated/config_flows.py
View file

@ -430,6 +430,7 @@ FLOWS = {
"xiaomi_ble",
"xiaomi_miio",
"yale_smart_alarm",
"yalexs_ble",
"yamaha_musiccast",
"yeelight",
"yolink",

3
requirements_all.txt
View file

@ -2499,6 +2499,9 @@ xs1-api-client==3.0.0
# homeassistant.components.yale_smart_alarm
yalesmartalarmclient==0.3.8
# homeassistant.components.yalexs_ble
yalexs-ble==1.1.2
# homeassistant.components.august
yalexs==1.2.1

3
requirements_test_all.txt
View file

@ -1694,6 +1694,9 @@ xmltodict==0.13.0
# homeassistant.components.yale_smart_alarm
yalesmartalarmclient==0.3.8
# homeassistant.components.yalexs_ble
yalexs-ble==1.1.2
# homeassistant.components.august
yalexs==1.2.1

67
tests/components/yalexs_ble/__init__.py Normal file
View file

@ -0,0 +1,67 @@
"""Tests for the Yale Access Bluetooth integration."""
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
YALE_ACCESS_LOCK_DISCOVERY_INFO = BluetoothServiceInfoBleak(
name="M1012LU",
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
manufacturer_data={
465: b"\x00\x00\xd1\xf0b;\xd8\x1dE\xd6\xba\xeeL\xdd]\xf5\xb2\xe9",
76: 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="M1012LU"),
advertisement=AdvertisementData(),
)
LOCK_DISCOVERY_INFO_UUID_ADDRESS = BluetoothServiceInfoBleak(
name="M1012LU",
address="61DE521B-F0BF-9F44-64D4-75BBE1738105",
rssi=-60,
manufacturer_data={
465: b"\x00\x00\xd1\xf0b;\xd8\x1dE\xd6\xba\xeeL\xdd]\xf5\xb2\xe9",
76: 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="M1012LU"),
advertisement=AdvertisementData(),
)
OLD_FIRMWARE_LOCK_DISCOVERY_INFO = BluetoothServiceInfoBleak(
name="Aug",
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
manufacturer_data={
465: b"\x00\x00\xd1\xf0b;\xd8\x1dE\xd6\xba\xeeL\xdd]\xf5\xb2\xe9",
76: 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=AdvertisementData(),
)
NOT_YALE_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=AdvertisementData(),
)

8
tests/components/yalexs_ble/conftest.py Normal file
View file

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

754
tests/components/yalexs_ble/test_config_flow.py Normal file
View file

@ -0,0 +1,754 @@
"""Test the Yale Access Bluetooth config flow."""
import asyncio
from unittest.mock import patch
from bleak import BleakError
from yalexs_ble import AuthError
from homeassistant import config_entries
from homeassistant.components.yalexs_ble.const import (
CONF_KEY,
CONF_LOCAL_NAME,
CONF_SLOT,
DOMAIN,
)
from homeassistant.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import (
LOCK_DISCOVERY_INFO_UUID_ADDRESS,
NOT_YALE_DISCOVERY_INFO,
OLD_FIRMWARE_LOCK_DISCOVERY_INFO,
YALE_ACCESS_LOCK_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.yalexs_ble.config_flow.async_discovered_service_info",
return_value=[NOT_YALE_DISCOVERY_INFO, YALE_ACCESS_LOCK_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.yalexs_ble.config_flow.PushLock.validate",
), patch(
"homeassistant.components.yalexs_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == YALE_ACCESS_LOCK_DISCOVERY_INFO.name
assert result2["data"] == {
CONF_LOCAL_NAME: YALE_ACCESS_LOCK_DISCOVERY_INFO.name,
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
}
assert result2["result"].unique_id == YALE_ACCESS_LOCK_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.yalexs_ble.config_flow.async_discovered_service_info",
return_value=[NOT_YALE_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_LOCAL_NAME: YALE_ACCESS_LOCK_DISCOVERY_INFO.name,
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
unique_id=YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.yalexs_ble.config_flow.async_discovered_service_info",
return_value=[YALE_ACCESS_LOCK_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_invalid_keys(hass: HomeAssistant) -> None:
"""Test user step with invalid keys tried first."""
with patch(
"homeassistant.components.yalexs_ble.config_flow.async_discovered_service_info",
return_value=[YALE_ACCESS_LOCK_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"] == {}
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "dog",
CONF_SLOT: 66,
},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {CONF_KEY: "invalid_key_format"}
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "qfd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
)
assert result3["type"] == FlowResultType.FORM
assert result3["step_id"] == "user"
assert result3["errors"] == {CONF_KEY: "invalid_key_format"}
result4 = await hass.config_entries.flow.async_configure(
result3["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 999,
},
)
assert result4["type"] == FlowResultType.FORM
assert result4["step_id"] == "user"
assert result4["errors"] == {CONF_SLOT: "invalid_key_index"}
with patch(
"homeassistant.components.yalexs_ble.config_flow.PushLock.validate",
), patch(
"homeassistant.components.yalexs_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result5 = await hass.config_entries.flow.async_configure(
result4["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
)
await hass.async_block_till_done()
assert result5["type"] == FlowResultType.CREATE_ENTRY
assert result5["title"] == YALE_ACCESS_LOCK_DISCOVERY_INFO.name
assert result5["data"] == {
CONF_LOCAL_NAME: YALE_ACCESS_LOCK_DISCOVERY_INFO.name,
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
}
assert result5["result"].unique_id == YALE_ACCESS_LOCK_DISCOVERY_INFO.address
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_step_cannot_connect(hass: HomeAssistant) -> None:
"""Test user step and we cannot connect."""
with patch(
"homeassistant.components.yalexs_ble.config_flow.async_discovered_service_info",
return_value=[YALE_ACCESS_LOCK_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.yalexs_ble.config_flow.PushLock.validate",
side_effect=BleakError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
)
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.yalexs_ble.config_flow.PushLock.validate",
), patch(
"homeassistant.components.yalexs_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == YALE_ACCESS_LOCK_DISCOVERY_INFO.name
assert result3["data"] == {
CONF_LOCAL_NAME: YALE_ACCESS_LOCK_DISCOVERY_INFO.name,
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
}
assert result3["result"].unique_id == YALE_ACCESS_LOCK_DISCOVERY_INFO.address
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_step_auth_exception(hass: HomeAssistant) -> None:
"""Test user step with an authentication exception."""
with patch(
"homeassistant.components.yalexs_ble.config_flow.async_discovered_service_info",
return_value=[YALE_ACCESS_LOCK_DISCOVERY_INFO, NOT_YALE_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.yalexs_ble.config_flow.PushLock.validate",
side_effect=AuthError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {CONF_KEY: "invalid_auth"}
with patch(
"homeassistant.components.yalexs_ble.config_flow.PushLock.validate",
), patch(
"homeassistant.components.yalexs_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == YALE_ACCESS_LOCK_DISCOVERY_INFO.name
assert result3["data"] == {
CONF_LOCAL_NAME: YALE_ACCESS_LOCK_DISCOVERY_INFO.name,
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
}
assert result3["result"].unique_id == YALE_ACCESS_LOCK_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.yalexs_ble.config_flow.async_discovered_service_info",
return_value=[NOT_YALE_DISCOVERY_INFO, YALE_ACCESS_LOCK_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.yalexs_ble.config_flow.PushLock.validate",
side_effect=RuntimeError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
)
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.yalexs_ble.config_flow.PushLock.validate",
), patch(
"homeassistant.components.yalexs_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == YALE_ACCESS_LOCK_DISCOVERY_INFO.name
assert result3["data"] == {
CONF_LOCAL_NAME: YALE_ACCESS_LOCK_DISCOVERY_INFO.name,
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
}
assert result3["result"].unique_id == YALE_ACCESS_LOCK_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=YALE_ACCESS_LOCK_DISCOVERY_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
with patch(
"homeassistant.components.yalexs_ble.config_flow.PushLock.validate",
), patch(
"homeassistant.components.yalexs_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == YALE_ACCESS_LOCK_DISCOVERY_INFO.name
assert result2["data"] == {
CONF_LOCAL_NAME: YALE_ACCESS_LOCK_DISCOVERY_INFO.name,
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
}
assert result2["result"].unique_id == YALE_ACCESS_LOCK_DISCOVERY_INFO.address
assert len(mock_setup_entry.mock_calls) == 1
async def test_integration_discovery_success(hass: HomeAssistant) -> None:
"""Test integration discovery step success path."""
with patch(
"homeassistant.components.yalexs_ble.util.async_process_advertisements",
return_value=YALE_ACCESS_LOCK_DISCOVERY_INFO,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
"name": "Front Door",
"address": YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
"key": "2fd51b8621c6a139eaffbedcb846b60f",
"slot": 66,
"serial": "M1XXX012LU",
},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "integration_discovery_confirm"
assert result["errors"] is None
with patch(
"homeassistant.components.yalexs_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Front Door"
assert result2["data"] == {
CONF_LOCAL_NAME: YALE_ACCESS_LOCK_DISCOVERY_INFO.name,
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
}
assert result2["result"].unique_id == YALE_ACCESS_LOCK_DISCOVERY_INFO.address
assert len(mock_setup_entry.mock_calls) == 1
async def test_integration_discovery_device_not_found(hass: HomeAssistant) -> None:
"""Test integration discovery when the device is not found."""
with patch(
"homeassistant.components.yalexs_ble.util.async_process_advertisements",
side_effect=asyncio.TimeoutError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
"name": "Front Door",
"address": YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
"key": "2fd51b8621c6a139eaffbedcb846b60f",
"slot": 66,
"serial": "M1XXX012LU",
},
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "no_devices_found"
async def test_integration_discovery_takes_precedence_over_bluetooth(
hass: HomeAssistant,
) -> None:
"""Test integration discovery dismisses bluetooth discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=YALE_ACCESS_LOCK_DISCOVERY_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
flows = [
flow
for flow in hass.config_entries.flow.async_progress()
if flow["handler"] == DOMAIN
]
assert len(flows) == 1
assert flows[0]["context"]["unique_id"] == YALE_ACCESS_LOCK_DISCOVERY_INFO.address
assert flows[0]["context"]["local_name"] == YALE_ACCESS_LOCK_DISCOVERY_INFO.name
with patch(
"homeassistant.components.yalexs_ble.util.async_process_advertisements",
return_value=YALE_ACCESS_LOCK_DISCOVERY_INFO,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
"name": "Front Door",
"address": YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
"key": "2fd51b8621c6a139eaffbedcb846b60f",
"slot": 66,
"serial": "M1XXX012LU",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "integration_discovery_confirm"
assert result["errors"] is None
# the bluetooth flow should get dismissed in favor
# of the integration discovery flow since the integration
# discovery flow will have the keys and the bluetooth
# flow will not
flows = [
flow
for flow in hass.config_entries.flow.async_progress()
if flow["handler"] == DOMAIN
]
assert len(flows) == 1
with patch(
"homeassistant.components.yalexs_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Front Door"
assert result2["data"] == {
CONF_LOCAL_NAME: YALE_ACCESS_LOCK_DISCOVERY_INFO.name,
CONF_ADDRESS: YALE_ACCESS_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
}
assert result2["result"].unique_id == YALE_ACCESS_LOCK_DISCOVERY_INFO.address
assert len(mock_setup_entry.mock_calls) == 1
flows = [
flow
for flow in hass.config_entries.flow.async_progress()
if flow["handler"] == DOMAIN
]
assert len(flows) == 0
async def test_integration_discovery_updates_key_unique_local_name(
hass: HomeAssistant,
) -> None:
"""Test integration discovery updates the key with a unique local name."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_LOCAL_NAME: LOCK_DISCOVERY_INFO_UUID_ADDRESS.name,
CONF_ADDRESS: "61DE521B-F0BF-9F44-64D4-75BBE1738105",
CONF_KEY: "5fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 11,
},
unique_id="61DE521B-F0BF-9F44-64D4-75BBE1738105",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.yalexs_ble.util.async_process_advertisements",
return_value=LOCK_DISCOVERY_INFO_UUID_ADDRESS,
), patch(
"homeassistant.components.yalexs_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
"name": "Front Door",
"address": "AA:BB:CC:DD:EE:FF",
"key": "2fd51b8621c6a139eaffbedcb846b60f",
"slot": 66,
"serial": "M1XXX012LU",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert entry.data[CONF_KEY] == "2fd51b8621c6a139eaffbedcb846b60f"
assert entry.data[CONF_SLOT] == 66
assert len(mock_setup_entry.mock_calls) == 1
async def test_integration_discovery_updates_key_without_unique_local_name(
hass: HomeAssistant,
) -> None:
"""Test integration discovery updates the key without a unique local name."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_LOCAL_NAME: OLD_FIRMWARE_LOCK_DISCOVERY_INFO.name,
CONF_ADDRESS: OLD_FIRMWARE_LOCK_DISCOVERY_INFO.address,
CONF_KEY: "5fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 11,
},
unique_id=OLD_FIRMWARE_LOCK_DISCOVERY_INFO.address,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.yalexs_ble.util.async_process_advertisements",
return_value=LOCK_DISCOVERY_INFO_UUID_ADDRESS,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
"name": "Front Door",
"address": OLD_FIRMWARE_LOCK_DISCOVERY_INFO.address,
"key": "2fd51b8621c6a139eaffbedcb846b60f",
"slot": 66,
"serial": "M1XXX012LU",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert entry.data[CONF_KEY] == "2fd51b8621c6a139eaffbedcb846b60f"
assert entry.data[CONF_SLOT] == 66
async def test_integration_discovery_takes_precedence_over_bluetooth_uuid_address(
hass: HomeAssistant,
) -> None:
"""Test integration discovery dismisses bluetooth discovery with a uuid address."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=LOCK_DISCOVERY_INFO_UUID_ADDRESS,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
flows = [
flow
for flow in hass.config_entries.flow.async_progress()
if flow["handler"] == DOMAIN
]
assert len(flows) == 1
assert flows[0]["context"]["unique_id"] == LOCK_DISCOVERY_INFO_UUID_ADDRESS.address
assert flows[0]["context"]["local_name"] == LOCK_DISCOVERY_INFO_UUID_ADDRESS.name
with patch(
"homeassistant.components.yalexs_ble.util.async_process_advertisements",
return_value=LOCK_DISCOVERY_INFO_UUID_ADDRESS,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
"name": "Front Door",
"address": "AA:BB:CC:DD:EE:FF",
"key": "2fd51b8621c6a139eaffbedcb846b60f",
"slot": 66,
"serial": "M1XXX012LU",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "integration_discovery_confirm"
assert result["errors"] is None
# the bluetooth flow should get dismissed in favor
# of the integration discovery flow since the integration
# discovery flow will have the keys and the bluetooth
# flow will not
flows = [
flow
for flow in hass.config_entries.flow.async_progress()
if flow["handler"] == DOMAIN
]
assert len(flows) == 1
with patch(
"homeassistant.components.yalexs_ble.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Front Door"
assert result2["data"] == {
CONF_LOCAL_NAME: LOCK_DISCOVERY_INFO_UUID_ADDRESS.name,
CONF_ADDRESS: LOCK_DISCOVERY_INFO_UUID_ADDRESS.address,
CONF_KEY: "2fd51b8621c6a139eaffbedcb846b60f",
CONF_SLOT: 66,
}
assert result2["result"].unique_id == OLD_FIRMWARE_LOCK_DISCOVERY_INFO.address
assert len(mock_setup_entry.mock_calls) == 1
flows = [
flow
for flow in hass.config_entries.flow.async_progress()
if flow["handler"] == DOMAIN
]
assert len(flows) == 0
async def test_integration_discovery_takes_precedence_over_bluetooth_non_unique_local_name(
hass: HomeAssistant,
) -> None:
"""Test integration discovery dismisses bluetooth discovery with a non unique local name."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=OLD_FIRMWARE_LOCK_DISCOVERY_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
flows = [
flow
for flow in hass.config_entries.flow.async_progress()
if flow["handler"] == DOMAIN
]
assert len(flows) == 1
assert flows[0]["context"]["unique_id"] == OLD_FIRMWARE_LOCK_DISCOVERY_INFO.address
assert flows[0]["context"]["local_name"] == OLD_FIRMWARE_LOCK_DISCOVERY_INFO.name
with patch(
"homeassistant.components.yalexs_ble.util.async_process_advertisements",
return_value=OLD_FIRMWARE_LOCK_DISCOVERY_INFO,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
"name": "Front Door",
"address": OLD_FIRMWARE_LOCK_DISCOVERY_INFO.address,
"key": "2fd51b8621c6a139eaffbedcb846b60f",
"slot": 66,
"serial": "M1XXX012LU",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "integration_discovery_confirm"
assert result["errors"] is None
# the bluetooth flow should get dismissed in favor
# of the integration discovery flow since the integration
# discovery flow will have the keys and the bluetooth
# flow will not
flows = [
flow
for flow in hass.config_entries.flow.async_progress()
if flow["handler"] == DOMAIN
]
assert len(flows) == 1