self-hosted/home-assistant-core
self-hosted/home-assistant-core
1
0
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-10-07 10:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support for encrypted BLE MiBeacon devices (#75677)

Browse source 
* Support for encrypted devices

* Sensor should use bindkey if available

* Error message if encryption fails

* Let mypy know this is always set by now

* Towards supporting encryption in step_user

* Add tests for the 4 new happy paths

* Add test coverage for failure cases

* Add strings

* Bump to 0.5.1. Legacy MiBeacon does not use an authentication token, so harder to detect incorrect key

* Add _title() helper

* Fix test after rebase

* Update homeassistant/components/xiaomi_ble/strings.json

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Remove unused lines

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Jc2k 2022-07-24 20:00:56 +01:00 committed by GitHub
parent f94a79b409
commit e18819c678
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 649 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

141
homeassistant/components/xiaomi_ble/config_flow.py
View file

@ -1,10 +1,12 @@
"""Config flow for Xiaomi Bluetooth integration."""
from __future__ import annotations
import dataclasses
from typing import Any
import voluptuous as vol
from xiaomi_ble import XiaomiBluetoothDeviceData as DeviceData
from xiaomi_ble.parser import EncryptionScheme
from homeassistant.components.bluetooth import (
BluetoothServiceInfo,
@ -17,6 +19,19 @@ from homeassistant.data_entry_flow import FlowResult
from .const import DOMAIN
@dataclasses.dataclass
class Discovery:
"""A discovered bluetooth device."""
title: str
discovery_info: BluetoothServiceInfo
device: DeviceData
def _title(discovery_info: BluetoothServiceInfo, device: DeviceData) -> str:
return device.title or device.get_device_name() or discovery_info.name
class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Xiaomi Bluetooth."""
@ -26,7 +41,7 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
"""Initialize the config flow."""
self._discovery_info: BluetoothServiceInfo | None = None
self._discovered_device: DeviceData | None = None
self._discovered_devices: dict[str, str] = {}
self._discovered_devices: dict[str, Discovery] = {}
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfo
@ -39,25 +54,101 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
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}
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()
return await self.async_step_bluetooth_confirm()
async def async_step_get_encryption_key_legacy(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Enter a legacy bindkey for a v2/v3 MiBeacon device."""
assert self._discovery_info
errors = {}
if user_input is not None:
bindkey = user_input["bindkey"]
if len(bindkey) != 24:
errors["bindkey"] = "expected_24_characters"
else:
device = DeviceData(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)
if device.bindkey_verified:
return self.async_create_entry(
title=self.context["title_placeholders"]["name"],
data={"bindkey": bindkey},
)
errors["bindkey"] = "decryption_failed"
return self.async_show_form(
step_id="get_encryption_key_legacy",
description_placeholders=self.context["title_placeholders"],
data_schema=vol.Schema({vol.Required("bindkey"): vol.All(str, vol.Strip)}),
errors=errors,
)
async def async_step_get_encryption_key_4_5(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Enter a bindkey for a v4/v5 MiBeacon device."""
assert self._discovery_info
errors = {}
if user_input is not None:
bindkey = user_input["bindkey"]
if len(bindkey) != 32:
errors["bindkey"] = "expected_32_characters"
else:
device = DeviceData(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)
if device.bindkey_verified:
return self.async_create_entry(
title=self.context["title_placeholders"]["name"],
data={"bindkey": bindkey},
)
errors["bindkey"] = "decryption_failed"
return self.async_show_form(
step_id="get_encryption_key_4_5",
description_placeholders=self.context["title_placeholders"],
data_schema=vol.Schema({vol.Required("bindkey"): vol.All(str, vol.Strip)}),
errors=errors,
)
async def async_step_bluetooth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm discovery."""
assert self._discovered_device is not None
device = self._discovered_device
assert self._discovery_info is not None
discovery_info = self._discovery_info
title = device.title or device.get_device_name() or discovery_info.name
if user_input is not None:
return self.async_create_entry(title=title, data={})
return self.async_create_entry(
title=self.context["title_placeholders"]["name"],
data={},
)
self._set_confirm_only()
placeholders = {"name": title}
self.context["title_placeholders"] = placeholders
return self.async_show_form(
step_id="bluetooth_confirm", description_placeholders=placeholders
step_id="bluetooth_confirm",
description_placeholders=self.context["title_placeholders"],
)
async def async_step_user(
@ -67,9 +158,19 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
if user_input is not None:
address = user_input[CONF_ADDRESS]
await self.async_set_unique_id(address, raise_on_progress=False)
return self.async_create_entry(
title=self._discovered_devices[address], data={}
)
discovery = self._discovered_devices[address]
if discovery.device.encryption_scheme == EncryptionScheme.MIBEACON_LEGACY:
self._discovery_info = discovery.discovery_info
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._discovery_info = discovery.discovery_info
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={})
current_addresses = self._async_current_ids()
for discovery_info in async_discovered_service_info(self.hass):
@ -78,16 +179,20 @@ class XiaomiConfigFlow(ConfigFlow, domain=DOMAIN):
continue
device = DeviceData()
if device.supported(discovery_info):
self._discovered_devices[address] = (
device.title or device.get_device_name() or discovery_info.name
self._discovered_devices[address] = Discovery(
title=_title(discovery_info, device),
discovery_info=discovery_info,
device=device,
)
if not self._discovered_devices:
return self.async_abort(reason="no_devices_found")
titles = {
address: discovery.title
for (address, discovery) in self._discovered_devices.items()
}
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_ADDRESS): vol.In(self._discovered_devices)}
),
data_schema=vol.Schema({vol.Required(CONF_ADDRESS): vol.In(titles)}),
)

2
homeassistant/components/xiaomi_ble/manifest.json
View file

@ -8,7 +8,7 @@
"service_uuid": "0000fe95-0000-1000-8000-00805f9b34fb"
}
],
"requirements": ["xiaomi-ble==0.2.0"],
"requirements": ["xiaomi-ble==0.5.1"],
"dependencies": ["bluetooth"],
"codeowners": ["@Jc2k", "@Ernst79"],
"iot_class": "local_push"

5
homeassistant/components/xiaomi_ble/sensor.py
View file

@ -158,7 +158,10 @@ async def async_setup_entry(
coordinator: PassiveBluetoothProcessorCoordinator = hass.data[DOMAIN][
entry.entry_id
]
data = XiaomiBluetoothDeviceData()
kwargs = {}
if bindkey := entry.data.get("bindkey"):
kwargs["bindkey"] = bytes.fromhex(bindkey)
data = XiaomiBluetoothDeviceData(**kwargs)
processor = PassiveBluetoothDataProcessor(
lambda service_info: sensor_update_to_bluetooth_data_update(
data.update(service_info)

17
homeassistant/components/xiaomi_ble/strings.json
View file

@ -10,12 +10,27 @@
},
"bluetooth_confirm": {
"description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]"
},
"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": {
"bindkey": "Bindkey"
}
},
"get_encryption_key_4_5": {
"description": "The sensor data broadcast by the sensor is encrypted. In order to decrypt it we need a 32 character hexadecimal bindkey.",
"data": {
"bindkey": "Bindkey"
}
}
},
"abort": {
"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%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"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."
}
}
}

2
requirements_all.txt
View file

@ -2477,7 +2477,7 @@ xbox-webapi==2.0.11
xboxapi==2.0.1
# homeassistant.components.xiaomi_ble
xiaomi-ble==0.2.0
xiaomi-ble==0.5.1
# homeassistant.components.knx
xknx==0.21.5

2
requirements_test_all.txt
View file

@ -1662,7 +1662,7 @@ wolf_smartset==0.1.11
xbox-webapi==2.0.11
# homeassistant.components.xiaomi_ble
xiaomi-ble==0.2.0
xiaomi-ble==0.5.1
# homeassistant.components.knx
xknx==0.21.5

24
tests/components/xiaomi_ble/__init__.py
View file

@ -37,6 +37,30 @@ MMC_T201_1_SERVICE_INFO = BluetoothServiceInfo(
source="local",
)
JTYJGD03MI_SERVICE_INFO = BluetoothServiceInfo(
name="JTYJGD03MI",
address="54:EF:44:E3:9C:BC",
rssi=-56,
manufacturer_data={},
service_data={
"0000fe95-0000-1000-8000-00805f9b34fb": b'XY\x97\td\xbc\x9c\xe3D\xefT" `\x88\xfd\x00\x00\x00\x00:\x14\x8f\xb3'
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
)
YLKG07YL_SERVICE_INFO = BluetoothServiceInfo(
name="YLKG07YL",
address="F8:24:41:C5:98:8B",
rssi=-56,
manufacturer_data={},
service_data={
"0000fe95-0000-1000-8000-00805f9b34fb": b"X0\xb6\x03\xd2\x8b\x98\xc5A$\xf8\xc3I\x14vu~\x00\x00\x00\x99",
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
)
def make_advertisement(address: str, payload: bytes) -> BluetoothServiceInfo:
"""Make a dummy advertisement."""

434
tests/components/xiaomi_ble/test_config_flow.py
View file

@ -7,9 +7,11 @@ from homeassistant.components.xiaomi_ble.const import DOMAIN
from homeassistant.data_entry_flow import FlowResultType
from . import (
JTYJGD03MI_SERVICE_INFO,
LYWSDCGQ_SERVICE_INFO,
MMC_T201_1_SERVICE_INFO,
NOT_SENSOR_PUSH_SERVICE_INFO,
YLKG07YL_SERVICE_INFO,
)
from tests.common import MockConfigEntry
@ -36,6 +38,187 @@ async def test_async_step_bluetooth_valid_device(hass):
assert result2["result"].unique_id == "00:81:F9:DD:6F:C1"
async def test_async_step_bluetooth_valid_device_legacy_encryption(hass):
"""Test discovery via bluetooth with a valid device, with legacy encryption."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=YLKG07YL_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "get_encryption_key_legacy"
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={"bindkey": "b853075158487ca39a5b5ea9"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "YLKG07YL"
assert result2["data"] == {"bindkey": "b853075158487ca39a5b5ea9"}
assert result2["result"].unique_id == "F8:24:41:C5:98:8B"
async def test_async_step_bluetooth_valid_device_legacy_encryption_wrong_key(hass):
"""Test discovery via bluetooth with a valid device, with legacy encryption and invalid key."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=YLKG07YL_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "get_encryption_key_legacy"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "aaaaaaaaaaaaaaaaaaaaaaaa"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "get_encryption_key_legacy"
assert result2["errors"]["bindkey"] == "decryption_failed"
# Test can finish flow
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={"bindkey": "b853075158487ca39a5b5ea9"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "YLKG07YL"
assert result2["data"] == {"bindkey": "b853075158487ca39a5b5ea9"}
assert result2["result"].unique_id == "F8:24:41:C5:98:8B"
async def test_async_step_bluetooth_valid_device_legacy_encryption_wrong_key_length(
hass,
):
"""Test discovery via bluetooth with a valid device, with legacy encryption and wrong key length."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=YLKG07YL_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "get_encryption_key_legacy"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "aaaaaaaaaaaaaaaaaaaaaaa"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "get_encryption_key_legacy"
assert result2["errors"]["bindkey"] == "expected_24_characters"
# Test can finish flow
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={"bindkey": "b853075158487ca39a5b5ea9"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "YLKG07YL"
assert result2["data"] == {"bindkey": "b853075158487ca39a5b5ea9"}
assert result2["result"].unique_id == "F8:24:41:C5:98:8B"
async def test_async_step_bluetooth_valid_device_v4_encryption(hass):
"""Test discovery via bluetooth with a valid device, with v4 encryption."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=JTYJGD03MI_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "get_encryption_key_4_5"
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={"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "JTYJGD03MI"
assert result2["data"] == {"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"}
assert result2["result"].unique_id == "54:EF:44:E3:9C:BC"
async def test_async_step_bluetooth_valid_device_v4_encryption_wrong_key(hass):
"""Test discovery via bluetooth with a valid device, with v4 encryption and wrong key."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=JTYJGD03MI_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "get_encryption_key_4_5"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "get_encryption_key_4_5"
assert result2["errors"]["bindkey"] == "decryption_failed"
# Test can finish flow
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={"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "JTYJGD03MI"
assert result2["data"] == {"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"}
assert result2["result"].unique_id == "54:EF:44:E3:9C:BC"
async def test_async_step_bluetooth_valid_device_v4_encryption_wrong_key_length(hass):
"""Test discovery via bluetooth with a valid device, with v4 encryption and wrong key length."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=JTYJGD03MI_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "get_encryption_key_4_5"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "5b51a7c91cde6707c9ef18fda143a58"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "get_encryption_key_4_5"
assert result2["errors"]["bindkey"] == "expected_32_characters"
# Test can finish flow
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={"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "JTYJGD03MI"
assert result2["data"] == {"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"}
assert result2["result"].unique_id == "54:EF:44:E3:9C:BC"
async def test_async_step_bluetooth_not_xiaomi(hass):
"""Test discovery via bluetooth not xiaomi."""
result = await hass.config_entries.flow.async_init(
@ -82,6 +265,257 @@ async def test_async_step_user_with_found_devices(hass):
assert result2["result"].unique_id == "58:2D:34:35:93:21"
async def test_async_step_user_with_found_devices_v4_encryption(hass):
"""Test setup from service info cache with devices found, with v4 encryption."""
with patch(
"homeassistant.components.xiaomi_ble.config_flow.async_discovered_service_info",
return_value=[JTYJGD03MI_SERVICE_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"
result1 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "54:EF:44:E3:9C:BC"},
)
assert result1["type"] == FlowResultType.FORM
assert result1["step_id"] == "get_encryption_key_4_5"
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={"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "JTYJGD03MI"
assert result2["data"] == {"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"}
assert result2["result"].unique_id == "54:EF:44:E3:9C:BC"
async def test_async_step_user_with_found_devices_v4_encryption_wrong_key(hass):
"""Test setup from service info cache with devices found, with v4 encryption and wrong key."""
# Get a list of devices
with patch(
"homeassistant.components.xiaomi_ble.config_flow.async_discovered_service_info",
return_value=[JTYJGD03MI_SERVICE_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"
# Pick a device
result1 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "54:EF:44:E3:9C:BC"},
)
assert result1["type"] == FlowResultType.FORM
assert result1["step_id"] == "get_encryption_key_4_5"
# Try an incorrect key
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "get_encryption_key_4_5"
assert result2["errors"]["bindkey"] == "decryption_failed"
# Check can still finish flow
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={"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "JTYJGD03MI"
assert result2["data"] == {"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"}
assert result2["result"].unique_id == "54:EF:44:E3:9C:BC"
async def test_async_step_user_with_found_devices_v4_encryption_wrong_key_length(hass):
"""Test setup from service info cache with devices found, with v4 encryption and wrong key length."""
# Get a list of devices
with patch(
"homeassistant.components.xiaomi_ble.config_flow.async_discovered_service_info",
return_value=[JTYJGD03MI_SERVICE_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"
# Select a single device
result1 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "54:EF:44:E3:9C:BC"},
)
assert result1["type"] == FlowResultType.FORM
assert result1["step_id"] == "get_encryption_key_4_5"
# Try an incorrect key
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "5b51a7c91cde6707c9ef1dfda143a58"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "get_encryption_key_4_5"
assert result2["errors"]["bindkey"] == "expected_32_characters"
# Check can still finish flow
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={"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "JTYJGD03MI"
assert result2["data"] == {"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"}
assert result2["result"].unique_id == "54:EF:44:E3:9C:BC"
async def test_async_step_user_with_found_devices_legacy_encryption(hass):
"""Test setup from service info cache with devices found, with legacy encryption."""
with patch(
"homeassistant.components.xiaomi_ble.config_flow.async_discovered_service_info",
return_value=[YLKG07YL_SERVICE_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"
result1 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "F8:24:41:C5:98:8B"},
)
assert result1["type"] == FlowResultType.FORM
assert result1["step_id"] == "get_encryption_key_legacy"
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={"bindkey": "b853075158487ca39a5b5ea9"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "YLKG07YL"
assert result2["data"] == {"bindkey": "b853075158487ca39a5b5ea9"}
assert result2["result"].unique_id == "F8:24:41:C5:98:8B"
async def test_async_step_user_with_found_devices_legacy_encryption_wrong_key(
hass,
):
"""Test setup from service info cache with devices found, with legacy encryption and wrong key."""
with patch(
"homeassistant.components.xiaomi_ble.config_flow.async_discovered_service_info",
return_value=[YLKG07YL_SERVICE_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"
result1 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "F8:24:41:C5:98:8B"},
)
assert result1["type"] == FlowResultType.FORM
assert result1["step_id"] == "get_encryption_key_legacy"
# Enter an incorrect code
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "aaaaaaaaaaaaaaaaaaaaaaaa"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "get_encryption_key_legacy"
assert result2["errors"]["bindkey"] == "decryption_failed"
# Check you can finish the flow
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={"bindkey": "b853075158487ca39a5b5ea9"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "YLKG07YL"
assert result2["data"] == {"bindkey": "b853075158487ca39a5b5ea9"}
assert result2["result"].unique_id == "F8:24:41:C5:98:8B"
async def test_async_step_user_with_found_devices_legacy_encryption_wrong_key_length(
hass,
):
"""Test setup from service info cache with devices found, with legacy encryption and wrong key length."""
with patch(
"homeassistant.components.xiaomi_ble.config_flow.async_discovered_service_info",
return_value=[YLKG07YL_SERVICE_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"
result1 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "F8:24:41:C5:98:8B"},
)
assert result1["type"] == FlowResultType.FORM
assert result1["step_id"] == "get_encryption_key_legacy"
# Enter an incorrect code
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "b85307518487ca39a5b5ea9"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "get_encryption_key_legacy"
assert result2["errors"]["bindkey"] == "expected_24_characters"
# Check you can finish the flow
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={"bindkey": "b853075158487ca39a5b5ea9"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "YLKG07YL"
assert result2["data"] == {"bindkey": "b853075158487ca39a5b5ea9"}
assert result2["result"].unique_id == "F8:24:41:C5:98:8B"
async def test_async_step_user_with_found_devices_already_setup(hass):
"""Test setup from service info cache with devices found."""
entry = MockConfigEntry(

45
tests/components/xiaomi_ble/test_sensor.py
View file

@ -130,3 +130,48 @@ async def test_xiaomi_HHCCJCY01(hass):
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_xiaomi_CGDK2(hass):
"""This device has encrypion so we need to retrieve its bindkey from the configentry."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="58:2D:34:12:20:89",
data={"bindkey": "a3bfe9853dd85a620debe3620caaa351"},
)
entry.add_to_hass(hass)
saved_callback = None
def _async_register_callback(_hass, _callback, _matcher):
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
saved_callback(
make_advertisement(
"58:2D:34:12:20:89",
b"XXo\x06\x07\x89 \x124-X_\x17m\xd5O\x02\x00\x00/\xa4S\xfa",
),
BluetoothChange.ADVERTISEMENT,
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
temp_sensor = hass.states.get("sensor.test_device_temperature")
temp_sensor_attribtes = temp_sensor.attributes
assert temp_sensor.state == "22.6"
assert temp_sensor_attribtes[ATTR_FRIENDLY_NAME] == "Test Device Temperature"
assert temp_sensor_attribtes[ATTR_UNIT_OF_MEASUREMENT] == "°C"
assert temp_sensor_attribtes[ATTR_STATE_CLASS] == "measurement"
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()