Add reauth flow to xiaomi_ble, fixes problem adding LYWSD03MMC (#76028)
Jc2k
GitHub
parent
27ed3d324f
commit
652a8e9e8a
|
|
@ -2,6 +2,7 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Mapping
|
||||
import dataclasses
|
||||
from typing import Any
|
||||
|
||||
|
|
@ -12,7 +13,7 @@ from xiaomi_ble.parser import EncryptionScheme
|
|||
from homeassistant.components import onboarding
|
||||
from homeassistant.components.bluetooth import (
|
||||
BluetoothScanningMode,
|
||||
BluetoothServiceInfoBleak,
|
||||
BluetoothServiceInfo,
|
||||
async_discovered_service_info,
|
||||
async_process_advertisements,
|
||||
)
|
||||
|
|
@ -31,11 +32,11 @@ class Discovery:
|
|||
"""A discovered bluetooth device."""
|
||||
|
||||
title: str
|
||||
discovery_info: BluetoothServiceInfoBleak
|
||||
discovery_info: BluetoothServiceInfo
|
||||
device: DeviceData
|
||||
|
||||
|
||||
def _title(discovery_info: BluetoothServiceInfoBleak, device: DeviceData) -> str:
|
||||
def _title(discovery_info: BluetoothServiceInfo, device: DeviceData) -> str:
|
||||
return device.title or device.get_device_name() or discovery_info.name
|
||||
|
||||
|
||||
|
|
@ -46,19 +47,19 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the config flow."""
|
||||
self._discovery_info: BluetoothServiceInfoBleak | None = None
|
||||
self._discovery_info: BluetoothServiceInfo | None = None
|
||||
self._discovered_device: DeviceData | None = None
|
||||
self._discovered_devices: dict[str, Discovery] = {}
|
||||
|
||||
async def _async_wait_for_full_advertisement(
|
||||
self, discovery_info: BluetoothServiceInfoBleak, device: DeviceData
|
||||
) -> BluetoothServiceInfoBleak:
|
||||
self, discovery_info: BluetoothServiceInfo, device: DeviceData
|
||||
) -> BluetoothServiceInfo:
|
||||
"""Sometimes first advertisement we receive is blank or incomplete. Wait until we get a useful one."""
|
||||
if not device.pending:
|
||||
return discovery_info
|
||||
|
||||
def _process_more_advertisements(
|
||||
service_info: BluetoothServiceInfoBleak,
|
||||
service_info: BluetoothServiceInfo,
|
||||
) -> bool:
|
||||
device.update(service_info)
|
||||
return not device.pending
|
||||
|
|
@ -72,7 +73,7 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
)
|
||||
|
||||
async def async_step_bluetooth(
|
||||
self, discovery_info: BluetoothServiceInfoBleak
|
||||
self, discovery_info: BluetoothServiceInfo
|
||||
) -> FlowResult:
|
||||
"""Handle the bluetooth discovery step."""
|
||||
await self.async_set_unique_id(discovery_info.address)
|
||||
|
|
@ -81,20 +82,21 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
if not device.supported(discovery_info):
|
||||
return self.async_abort(reason="not_supported")
|
||||
|
||||
title = _title(discovery_info, device)
|
||||
self.context["title_placeholders"] = {"name": title}
|
||||
|
||||
self._discovered_device = device
|
||||
|
||||
# Wait until we have received enough information about this device to detect its encryption type
|
||||
try:
|
||||
discovery_info = await self._async_wait_for_full_advertisement(
|
||||
self._discovery_info = await self._async_wait_for_full_advertisement(
|
||||
discovery_info, device
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
# If we don't see a valid packet within the timeout then this device is not supported.
|
||||
return self.async_abort(reason="not_supported")
|
||||
|
||||
self._discovery_info = discovery_info
|
||||
self._discovered_device = device
|
||||
|
||||
title = _title(discovery_info, device)
|
||||
self.context["title_placeholders"] = {"name": title}
|
||||
# This device might have a really long advertising interval
|
||||
# So create a config entry for it, and if we discover it has encryption later
|
||||
# We can do a reauth
|
||||
return await self.async_step_confirm_slow()
|
||||
|
||||
if device.encryption_scheme == EncryptionScheme.MIBEACON_LEGACY:
|
||||
return await self.async_step_get_encryption_key_legacy()
|
||||
|
|
@ -107,6 +109,8 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
) -> FlowResult:
|
||||
"""Enter a legacy bindkey for a v2/v3 MiBeacon device."""
|
||||
assert self._discovery_info
|
||||
assert self._discovered_device
|
||||
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
|
|
@ -115,18 +119,15 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
if len(bindkey) != 24:
|
||||
errors["bindkey"] = "expected_24_characters"
|
||||
else:
|
||||
device = DeviceData(bindkey=bytes.fromhex(bindkey))
|
||||
self._discovered_device.bindkey = bytes.fromhex(bindkey)
|
||||
|
||||
# If we got this far we already know supported will
|
||||
# return true so we don't bother checking that again
|
||||
# We just want to retry the decryption
|
||||
device.supported(self._discovery_info)
|
||||
self._discovered_device.supported(self._discovery_info)
|
||||
|
||||
if device.bindkey_verified:
|
||||
return self.async_create_entry(
|
||||
title=self.context["title_placeholders"]["name"],
|
||||
data={"bindkey": bindkey},
|
||||
)
|
||||
if self._discovered_device.bindkey_verified:
|
||||
return self._async_get_or_create_entry(bindkey)
|
||||
|
||||
errors["bindkey"] = "decryption_failed"
|
||||
|
||||
|
|
@ -142,6 +143,7 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
) -> FlowResult:
|
||||
"""Enter a bindkey for a v4/v5 MiBeacon device."""
|
||||
assert self._discovery_info
|
||||
assert self._discovered_device
|
||||
|
||||
errors = {}
|
||||
|
||||
|
|
@ -151,18 +153,15 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
if len(bindkey) != 32:
|
||||
errors["bindkey"] = "expected_32_characters"
|
||||
else:
|
||||
device = DeviceData(bindkey=bytes.fromhex(bindkey))
|
||||
self._discovered_device.bindkey = bytes.fromhex(bindkey)
|
||||
|
||||
# If we got this far we already know supported will
|
||||
# return true so we don't bother checking that again
|
||||
# We just want to retry the decryption
|
||||
device.supported(self._discovery_info)
|
||||
self._discovered_device.supported(self._discovery_info)
|
||||
|
||||
if device.bindkey_verified:
|
||||
return self.async_create_entry(
|
||||
title=self.context["title_placeholders"]["name"],
|
||||
data={"bindkey": bindkey},
|
||||
)
|
||||
if self._discovered_device.bindkey_verified:
|
||||
return self._async_get_or_create_entry(bindkey)
|
||||
|
||||
errors["bindkey"] = "decryption_failed"
|
||||
|
||||
|
|
@ -178,10 +177,7 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
) -> FlowResult:
|
||||
"""Confirm discovery."""
|
||||
if user_input is not None or not onboarding.async_is_onboarded(self.hass):
|
||||
return self.async_create_entry(
|
||||
title=self.context["title_placeholders"]["name"],
|
||||
data={},
|
||||
)
|
||||
return self._async_get_or_create_entry()
|
||||
|
||||
self._set_confirm_only()
|
||||
return self.async_show_form(
|
||||
|
|
@ -189,6 +185,19 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
description_placeholders=self.context["title_placeholders"],
|
||||
)
|
||||
|
||||
async def async_step_confirm_slow(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Ack that device is slow."""
|
||||
if user_input is not None or not onboarding.async_is_onboarded(self.hass):
|
||||
return self._async_get_or_create_entry()
|
||||
|
||||
self._set_confirm_only()
|
||||
return self.async_show_form(
|
||||
step_id="confirm_slow",
|
||||
description_placeholders=self.context["title_placeholders"],
|
||||
)
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
|
|
@ -198,24 +207,28 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
await self.async_set_unique_id(address, raise_on_progress=False)
|
||||
discovery = self._discovered_devices[address]
|
||||
|
||||
self.context["title_placeholders"] = {"name": discovery.title}
|
||||
|
||||
# Wait until we have received enough information about this device to detect its encryption type
|
||||
try:
|
||||
self._discovery_info = await self._async_wait_for_full_advertisement(
|
||||
discovery.discovery_info, discovery.device
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
# If we don't see a valid packet within the timeout then this device is not supported.
|
||||
return self.async_abort(reason="not_supported")
|
||||
# This device might have a really long advertising interval
|
||||
# So create a config entry for it, and if we discover it has encryption later
|
||||
# We can do a reauth
|
||||
return await self.async_step_confirm_slow()
|
||||
|
||||
self._discovered_device = discovery.device
|
||||
|
||||
if discovery.device.encryption_scheme == EncryptionScheme.MIBEACON_LEGACY:
|
||||
self.context["title_placeholders"] = {"name": discovery.title}
|
||||
return await self.async_step_get_encryption_key_legacy()
|
||||
|
||||
if discovery.device.encryption_scheme == EncryptionScheme.MIBEACON_4_5:
|
||||
self.context["title_placeholders"] = {"name": discovery.title}
|
||||
return await self.async_step_get_encryption_key_4_5()
|
||||
|
||||
return self.async_create_entry(title=discovery.title, data={})
|
||||
return self._async_get_or_create_entry()
|
||||
|
||||
current_addresses = self._async_current_ids()
|
||||
for discovery_info in async_discovered_service_info(self.hass):
|
||||
|
|
@ -241,3 +254,46 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
step_id="user",
|
||||
data_schema=vol.Schema({vol.Required(CONF_ADDRESS): vol.In(titles)}),
|
||||
)
|
||||
|
||||
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
|
||||
"""Handle a flow initialized by a reauth event."""
|
||||
entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
|
||||
assert entry is not None
|
||||
|
||||
device: DeviceData = self.context["device"]
|
||||
self._discovered_device = device
|
||||
|
||||
self._discovery_info = device.last_service_info
|
||||
|
||||
if device.encryption_scheme == EncryptionScheme.MIBEACON_LEGACY:
|
||||
return await self.async_step_get_encryption_key_legacy()
|
||||
|
||||
if device.encryption_scheme == EncryptionScheme.MIBEACON_4_5:
|
||||
return await self.async_step_get_encryption_key_4_5()
|
||||
|
||||
# Otherwise there wasn't actually encryption so abort
|
||||
return self.async_abort(reason="reauth_successful")
|
||||
|
||||
def _async_get_or_create_entry(self, bindkey=None):
|
||||
data = {}
|
||||
|
||||
if bindkey:
|
||||
data["bindkey"] = bindkey
|
||||
|
||||
if entry_id := self.context.get("entry_id"):
|
||||
entry = self.hass.config_entries.async_get_entry(entry_id)
|
||||
assert entry is not None
|
||||
|
||||
self.hass.config_entries.async_update_entry(entry, data=data)
|
||||
|
||||
# Reload the config entry to notify of updated config
|
||||
self.hass.async_create_task(
|
||||
self.hass.config_entries.async_reload(entry.entry_id)
|
||||
)
|
||||
|
||||
return self.async_abort(reason="reauth_successful")
|
||||
|
||||
return self.async_create_entry(
|
||||
title=self.context["title_placeholders"]["name"],
|
||||
data=data,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@
|
|||
"service_data_uuid": "0000fe95-0000-1000-8000-00805f9b34fb"
|
||||
}
|
||||
],
|
||||
"requirements": ["xiaomi-ble==0.6.2"],
|
||||
"requirements": ["xiaomi-ble==0.6.4"],
|
||||
"dependencies": ["bluetooth"],
|
||||
"codeowners": ["@Jc2k", "@Ernst79"],
|
||||
"iot_class": "local_push"
|
||||
|
|
|
|||
|
|
@ -11,8 +11,10 @@ from xiaomi_ble import (
|
|||
Units,
|
||||
XiaomiBluetoothDeviceData,
|
||||
)
|
||||
from xiaomi_ble.parser import EncryptionScheme
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
|
||||
from homeassistant.components.bluetooth.passive_update_processor import (
|
||||
PassiveBluetoothDataProcessor,
|
||||
PassiveBluetoothDataUpdate,
|
||||
|
|
@ -163,6 +165,45 @@ def sensor_update_to_bluetooth_data_update(
|
|||
)
|
||||
|
||||
|
||||
def process_service_info(
|
||||
hass: HomeAssistant,
|
||||
entry: config_entries.ConfigEntry,
|
||||
data: XiaomiBluetoothDeviceData,
|
||||
service_info: BluetoothServiceInfoBleak,
|
||||
) -> PassiveBluetoothDataUpdate:
|
||||
"""Process a BluetoothServiceInfoBleak, running side effects and returning sensor data."""
|
||||
update = data.update(service_info)
|
||||
|
||||
# If device isn't pending we know it has seen at least one broadcast with a payload
|
||||
# If that payload was encrypted and the bindkey was not verified then we need to reauth
|
||||
if (
|
||||
not data.pending
|
||||
and data.encryption_scheme != EncryptionScheme.NONE
|
||||
and not data.bindkey_verified
|
||||
):
|
||||
flow_context = {
|
||||
"source": config_entries.SOURCE_REAUTH,
|
||||
"entry_id": entry.entry_id,
|
||||
"title_placeholders": {"name": entry.title},
|
||||
"unique_id": entry.unique_id,
|
||||
"device": data,
|
||||
}
|
||||
|
||||
for flow in hass.config_entries.flow.async_progress_by_handler(DOMAIN):
|
||||
if flow["context"] == flow_context:
|
||||
break
|
||||
else:
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context=flow_context,
|
||||
data=entry.data,
|
||||
)
|
||||
)
|
||||
|
||||
return sensor_update_to_bluetooth_data_update(update)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: config_entries.ConfigEntry,
|
||||
|
|
@ -177,9 +218,7 @@ async def async_setup_entry(
|
|||
kwargs["bindkey"] = bytes.fromhex(bindkey)
|
||||
data = XiaomiBluetoothDeviceData(**kwargs)
|
||||
processor = PassiveBluetoothDataProcessor(
|
||||
lambda service_info: sensor_update_to_bluetooth_data_update(
|
||||
data.update(service_info)
|
||||
)
|
||||
lambda service_info: process_service_info(hass, entry, data, service_info)
|
||||
)
|
||||
entry.async_on_unload(
|
||||
processor.async_add_entities_listener(
|
||||
|
|
|
|||
|
|
@ -11,6 +11,9 @@
|
|||
"bluetooth_confirm": {
|
||||
"description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]"
|
||||
},
|
||||
"slow_confirm": {
|
||||
"description": "There hasn't been a broadcast from this device in the last minute so we aren't sure if this device uses encryption or not. This may be because the device uses a slow broadcast interval. Confirm to add this device anyway, then the next time a broadcast is received you will be prompted to enter its bindkey if its needed."
|
||||
},
|
||||
"get_encryption_key_legacy": {
|
||||
"description": "The sensor data broadcast by the sensor is encrypted. In order to decrypt it we need a 24 character hexadecimal bindkey.",
|
||||
"data": {
|
||||
|
|
@ -25,6 +28,7 @@
|
|||
}
|
||||
},
|
||||
"abort": {
|
||||
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
|
||||
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
|
||||
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
|
|
|
|||
|
|
@ -6,7 +6,8 @@
|
|||
"decryption_failed": "The provided bindkey did not work, sensor data could not be decrypted. Please check it and try again.",
|
||||
"expected_24_characters": "Expected a 24 character hexadecimal bindkey.",
|
||||
"expected_32_characters": "Expected a 32 character hexadecimal bindkey.",
|
||||
"no_devices_found": "No devices found on the network"
|
||||
"no_devices_found": "No devices found on the network",
|
||||
"reauth_successful": "Re-authentication was successful"
|
||||
},
|
||||
"flow_title": "{name}",
|
||||
"step": {
|
||||
|
|
@ -25,6 +26,9 @@
|
|||
},
|
||||
"description": "The sensor data broadcast by the sensor is encrypted. In order to decrypt it we need a 24 character hexadecimal bindkey."
|
||||
},
|
||||
"slow_confirm": {
|
||||
"description": "There hasn't been a broadcast from this device in the last minute so we aren't sure if this device uses encryption or not. This may be because the device uses a slow broadcast interval. Confirm to add this device anyway, then the next time a broadcast is received you will be prompted to enter its bindkey if its needed."
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"address": "Device"
|
||||
|
|
|
|||
|
|
@ -2473,7 +2473,7 @@ xbox-webapi==2.0.11
|
|||
xboxapi==2.0.1
|
||||
|
||||
# homeassistant.components.xiaomi_ble
|
||||
xiaomi-ble==0.6.2
|
||||
xiaomi-ble==0.6.4
|
||||
|
||||
# homeassistant.components.knx
|
||||
xknx==0.22.1
|
||||
|
|
|
|||
|
|
@ -1665,7 +1665,7 @@ wolf_smartset==0.1.11
|
|||
xbox-webapi==2.0.11
|
||||
|
||||
# homeassistant.components.xiaomi_ble
|
||||
xiaomi-ble==0.6.2
|
||||
xiaomi-ble==0.6.4
|
||||
|
||||
# homeassistant.components.knx
|
||||
xknx==0.22.1
|
||||
|
|
|
|||
|
|
@ -3,7 +3,10 @@
|
|||
import asyncio
|
||||
from unittest.mock import patch
|
||||
|
||||
from xiaomi_ble import XiaomiBluetoothDeviceData as DeviceData
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.bluetooth import BluetoothChange
|
||||
from homeassistant.components.xiaomi_ble.const import DOMAIN
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
|
|
@ -52,8 +55,19 @@ async def test_async_step_bluetooth_valid_device_but_missing_payload(hass):
|
|||
context={"source": config_entries.SOURCE_BLUETOOTH},
|
||||
data=MISSING_PAYLOAD_ENCRYPTED,
|
||||
)
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "not_supported"
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "confirm_slow"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.xiaomi_ble.async_setup_entry", return_value=True
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={}
|
||||
)
|
||||
assert result2["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result2["title"] == "LYWSD02MMC"
|
||||
assert result2["data"] == {}
|
||||
assert result2["result"].unique_id == "A4:C1:38:56:53:84"
|
||||
|
||||
|
||||
async def test_async_step_bluetooth_valid_device_but_missing_payload_then_full(hass):
|
||||
|
|
@ -318,6 +332,24 @@ async def test_async_step_user_no_devices_found(hass):
|
|||
assert result["reason"] == "no_devices_found"
|
||||
|
||||
|
||||
async def test_async_step_user_no_devices_found_2(hass):
|
||||
"""
|
||||
Test setup from service info cache with no devices found.
|
||||
|
||||
This variant tests with a non-Xiaomi device known to us.
|
||||
"""
|
||||
with patch(
|
||||
"homeassistant.components.xiaomi_ble.config_flow.async_discovered_service_info",
|
||||
return_value=[NOT_SENSOR_PUSH_SERVICE_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_devices_found"
|
||||
|
||||
|
||||
async def test_async_step_user_with_found_devices(hass):
|
||||
"""Test setup from service info cache with devices found."""
|
||||
with patch(
|
||||
|
|
@ -363,8 +395,19 @@ async def test_async_step_user_short_payload(hass):
|
|||
result["flow_id"],
|
||||
user_input={"address": "A4:C1:38:56:53:84"},
|
||||
)
|
||||
assert result2["type"] == FlowResultType.ABORT
|
||||
assert result2["reason"] == "not_supported"
|
||||
assert result2["type"] == FlowResultType.FORM
|
||||
assert result2["step_id"] == "confirm_slow"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.xiaomi_ble.async_setup_entry", return_value=True
|
||||
):
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={}
|
||||
)
|
||||
assert result3["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result3["title"] == "LYWSD02MMC"
|
||||
assert result3["data"] == {}
|
||||
assert result3["result"].unique_id == "A4:C1:38:56:53:84"
|
||||
|
||||
|
||||
async def test_async_step_user_short_payload_then_full(hass):
|
||||
|
|
@ -755,3 +798,245 @@ async def test_async_step_user_takes_precedence_over_discovery(hass):
|
|||
|
||||
# Verify the original one was aborted
|
||||
assert not hass.config_entries.flow.async_progress(DOMAIN)
|
||||
|
||||
|
||||
async def test_async_step_reauth_legacy(hass):
|
||||
"""Test reauth with a legacy key."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id="F8:24:41:C5:98:8B",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
saved_callback = None
|
||||
|
||||
def _async_register_callback(_hass, _callback, _matcher, _mode):
|
||||
nonlocal saved_callback
|
||||
saved_callback = _callback
|
||||
return lambda: None
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.bluetooth.update_coordinator.async_register_callback",
|
||||
_async_register_callback,
|
||||
):
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_all()) == 0
|
||||
|
||||
# WARNING: This test data is synthetic, rather than captured from a real device
|
||||
# obj type is 0x1310, payload len is 0x2 and payload is 0x6000
|
||||
saved_callback(
|
||||
make_advertisement(
|
||||
"F8:24:41:C5:98:8B",
|
||||
b"X0\xb6\x03\xd2\x8b\x98\xc5A$\xf8\xc3I\x14vu~\x00\x00\x00\x99",
|
||||
),
|
||||
BluetoothChange.ADVERTISEMENT,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
results = hass.config_entries.flow.async_progress()
|
||||
assert len(results) == 1
|
||||
result = results[0]
|
||||
|
||||
assert result["step_id"] == "get_encryption_key_legacy"
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={"bindkey": "b853075158487ca39a5b5ea9"},
|
||||
)
|
||||
assert result2["type"] == FlowResultType.ABORT
|
||||
assert result2["reason"] == "reauth_successful"
|
||||
|
||||
|
||||
async def test_async_step_reauth_legacy_wrong_key(hass):
|
||||
"""Test reauth with a bad legacy key, and that we can recover."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id="F8:24:41:C5:98:8B",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
saved_callback = None
|
||||
|
||||
def _async_register_callback(_hass, _callback, _matcher, _mode):
|
||||
nonlocal saved_callback
|
||||
saved_callback = _callback
|
||||
return lambda: None
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.bluetooth.update_coordinator.async_register_callback",
|
||||
_async_register_callback,
|
||||
):
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_all()) == 0
|
||||
|
||||
# WARNING: This test data is synthetic, rather than captured from a real device
|
||||
# obj type is 0x1310, payload len is 0x2 and payload is 0x6000
|
||||
saved_callback(
|
||||
make_advertisement(
|
||||
"F8:24:41:C5:98:8B",
|
||||
b"X0\xb6\x03\xd2\x8b\x98\xc5A$\xf8\xc3I\x14vu~\x00\x00\x00\x99",
|
||||
),
|
||||
BluetoothChange.ADVERTISEMENT,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
results = hass.config_entries.flow.async_progress()
|
||||
assert len(results) == 1
|
||||
result = results[0]
|
||||
|
||||
assert result["step_id"] == "get_encryption_key_legacy"
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={"bindkey": "b85307515a487ca39a5b5ea9"},
|
||||
)
|
||||
assert result2["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "get_encryption_key_legacy"
|
||||
assert result2["errors"]["bindkey"] == "decryption_failed"
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={"bindkey": "b853075158487ca39a5b5ea9"},
|
||||
)
|
||||
assert result2["type"] == FlowResultType.ABORT
|
||||
assert result2["reason"] == "reauth_successful"
|
||||
|
||||
|
||||
async def test_async_step_reauth_v4(hass):
|
||||
"""Test reauth with a v4 key."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id="54:EF:44:E3:9C:BC",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
saved_callback = None
|
||||
|
||||
def _async_register_callback(_hass, _callback, _matcher, _mode):
|
||||
nonlocal saved_callback
|
||||
saved_callback = _callback
|
||||
return lambda: None
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.bluetooth.update_coordinator.async_register_callback",
|
||||
_async_register_callback,
|
||||
):
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_all()) == 0
|
||||
|
||||
# WARNING: This test data is synthetic, rather than captured from a real device
|
||||
# obj type is 0x1310, payload len is 0x2 and payload is 0x6000
|
||||
saved_callback(
|
||||
make_advertisement(
|
||||
"54:EF:44:E3:9C:BC",
|
||||
b"XY\x97\tf\xbc\x9c\xe3D\xefT\x01" b"\x08\x12\x05\x00\x00\x00q^\xbe\x90",
|
||||
),
|
||||
BluetoothChange.ADVERTISEMENT,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
results = hass.config_entries.flow.async_progress()
|
||||
assert len(results) == 1
|
||||
result = results[0]
|
||||
|
||||
assert result["step_id"] == "get_encryption_key_4_5"
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"},
|
||||
)
|
||||
assert result2["type"] == FlowResultType.ABORT
|
||||
assert result2["reason"] == "reauth_successful"
|
||||
|
||||
|
||||
async def test_async_step_reauth_v4_wrong_key(hass):
|
||||
"""Test reauth for v4 with a bad key, and that we can recover."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id="54:EF:44:E3:9C:BC",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
saved_callback = None
|
||||
|
||||
def _async_register_callback(_hass, _callback, _matcher, _mode):
|
||||
nonlocal saved_callback
|
||||
saved_callback = _callback
|
||||
return lambda: None
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.bluetooth.update_coordinator.async_register_callback",
|
||||
_async_register_callback,
|
||||
):
|
||||
assert await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_all()) == 0
|
||||
|
||||
# WARNING: This test data is synthetic, rather than captured from a real device
|
||||
# obj type is 0x1310, payload len is 0x2 and payload is 0x6000
|
||||
saved_callback(
|
||||
make_advertisement(
|
||||
"54:EF:44:E3:9C:BC",
|
||||
b"XY\x97\tf\xbc\x9c\xe3D\xefT\x01" b"\x08\x12\x05\x00\x00\x00q^\xbe\x90",
|
||||
),
|
||||
BluetoothChange.ADVERTISEMENT,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
results = hass.config_entries.flow.async_progress()
|
||||
assert len(results) == 1
|
||||
result = results[0]
|
||||
|
||||
assert result["step_id"] == "get_encryption_key_4_5"
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={"bindkey": "5b51a7c91cde6707c9ef18dada143a58"},
|
||||
)
|
||||
assert result2["type"] == FlowResultType.FORM
|
||||
assert result2["step_id"] == "get_encryption_key_4_5"
|
||||
assert result2["errors"]["bindkey"] == "decryption_failed"
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"},
|
||||
)
|
||||
assert result2["type"] == FlowResultType.ABORT
|
||||
assert result2["reason"] == "reauth_successful"
|
||||
|
||||
|
||||
async def test_async_step_reauth_abort_early(hass):
|
||||
"""
|
||||
Test we can abort the reauth if there is no encryption.
|
||||
|
||||
(This can't currently happen in practice).
|
||||
"""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id="54:EF:44:E3:9C:BC",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
device = DeviceData()
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={
|
||||
"source": config_entries.SOURCE_REAUTH,
|
||||
"entry_id": entry.entry_id,
|
||||
"title_placeholders": {"name": entry.title},
|
||||
"unique_id": entry.unique_id,
|
||||
"device": device,
|
||||
},
|
||||
data=entry.data,
|
||||
)
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "reauth_successful"
|
||||
|
|
|
|||
Loading…
Reference in New Issue