diff --git a/homeassistant/components/xiaomi_ble/config_flow.py b/homeassistant/components/xiaomi_ble/config_flow.py index 8f478442d6a..8b3ec22def7 100644 --- a/homeassistant/components/xiaomi_ble/config_flow.py +++ b/homeassistant/components/xiaomi_ble/config_flow.py @@ -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)}), ) diff --git a/homeassistant/components/xiaomi_ble/manifest.json b/homeassistant/components/xiaomi_ble/manifest.json index 17e22accd6d..457eac60407 100644 --- a/homeassistant/components/xiaomi_ble/manifest.json +++ b/homeassistant/components/xiaomi_ble/manifest.json @@ -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" diff --git a/homeassistant/components/xiaomi_ble/sensor.py b/homeassistant/components/xiaomi_ble/sensor.py index 50cbb0f66cf..9cdf661f5ad 100644 --- a/homeassistant/components/xiaomi_ble/sensor.py +++ b/homeassistant/components/xiaomi_ble/sensor.py @@ -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) diff --git a/homeassistant/components/xiaomi_ble/strings.json b/homeassistant/components/xiaomi_ble/strings.json index 7111626cca1..e12a15a0671 100644 --- a/homeassistant/components/xiaomi_ble/strings.json +++ b/homeassistant/components/xiaomi_ble/strings.json @@ -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." } } } diff --git a/requirements_all.txt b/requirements_all.txt index d6a4201cfd5..b078bf9d4d1 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -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 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index ae40622f221..269a2323760 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -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 diff --git a/tests/components/xiaomi_ble/__init__.py b/tests/components/xiaomi_ble/__init__.py index 80ec2f19989..a6269a02d12 100644 --- a/tests/components/xiaomi_ble/__init__.py +++ b/tests/components/xiaomi_ble/__init__.py @@ -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.""" diff --git a/tests/components/xiaomi_ble/test_config_flow.py b/tests/components/xiaomi_ble/test_config_flow.py index f99cbd21296..fc625bdf7ec 100644 --- a/tests/components/xiaomi_ble/test_config_flow.py +++ b/tests/components/xiaomi_ble/test_config_flow.py @@ -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( diff --git a/tests/components/xiaomi_ble/test_sensor.py b/tests/components/xiaomi_ble/test_sensor.py index 74a4fe65131..b95ea37311e 100644 --- a/tests/components/xiaomi_ble/test_sensor.py +++ b/tests/components/xiaomi_ble/test_sensor.py @@ -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()