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

Add support for BThome (#77224) 

* Add BThome BLE

* Update BThome to latest version

* 0.3.4

* Rename to bthome 2

* Fix uuids not being found

* Make energy a total increasing state class

* Change unit of measurement of VOC

* Use short identifier

* Fix the reauth flow

* Bump bthome_ble

* Parameterize sensor tests

* Remove Move function to parameter

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
pull/77417/head
Ernst Klamer 2022-08-27 15:25:29 +02:00 committed by GitHub
parent a6770f8b03
commit b2e958292c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 1596 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS
View File

@ -159,6 +159,8 @@ build.json @home-assistant/supervisor
/homeassistant/components/bsblan/ @liudger
/tests/components/bsblan/ @liudger
/homeassistant/components/bt_smarthub/ @jxwolstenholme
/homeassistant/components/bthome/ @Ernst79
/tests/components/bthome/ @Ernst79
/homeassistant/components/buienradar/ @mjj4791 @ties @Robbie1221
/tests/components/buienradar/ @mjj4791 @ties @Robbie1221
/homeassistant/components/button/ @home-assistant/core

76
homeassistant/components/bthome/__init__.py Normal file
View File

@ -0,0 +1,76 @@
"""The BThome Bluetooth integration."""
from __future__ import annotations
import logging
from bthome_ble import BThomeBluetoothDeviceData, SensorUpdate
from bthome_ble.parser import EncryptionScheme
from homeassistant.components.bluetooth import (
BluetoothScanningMode,
BluetoothServiceInfoBleak,
)
from homeassistant.components.bluetooth.passive_update_processor import (
PassiveBluetoothProcessorCoordinator,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .const import DOMAIN
PLATFORMS: list[Platform] = [Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
def process_service_info(
hass: HomeAssistant,
entry: ConfigEntry,
data: BThomeBluetoothDeviceData,
service_info: BluetoothServiceInfoBleak,
) -> SensorUpdate:
"""Process a BluetoothServiceInfoBleak, running side effects and returning sensor data."""
update = data.update(service_info)
# If that payload was encrypted and the bindkey was not verified then we need to reauth
if data.encryption_scheme != EncryptionScheme.NONE and not data.bindkey_verified:
entry.async_start_reauth(hass, data={"device": data})
return update
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up BThome Bluetooth from a config entry."""
address = entry.unique_id
assert address is not None
kwargs = {}
if bindkey := entry.data.get("bindkey"):
kwargs["bindkey"] = bytes.fromhex(bindkey)
data = BThomeBluetoothDeviceData(**kwargs)
coordinator = hass.data.setdefault(DOMAIN, {})[
entry.entry_id
] = PassiveBluetoothProcessorCoordinator(
hass,
_LOGGER,
address=address,
mode=BluetoothScanningMode.PASSIVE,
update_method=lambda service_info: process_service_info(
hass, entry, data, service_info
),
connectable=False,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(
coordinator.async_start()
) # only start after all platforms have had a chance to subscribe
return True
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

197
homeassistant/components/bthome/config_flow.py Normal file
View File

@ -0,0 +1,197 @@
"""Config flow for BThome Bluetooth integration."""
from __future__ import annotations
from collections.abc import Mapping
import dataclasses
from typing import Any
from bthome_ble import BThomeBluetoothDeviceData as DeviceData
from bthome_ble.parser import EncryptionScheme
import voluptuous as vol
from homeassistant.components import onboarding
from homeassistant.components.bluetooth import (
BluetoothServiceInfo,
async_discovered_service_info,
)
from homeassistant.config_entries import ConfigFlow
from homeassistant.const import CONF_ADDRESS
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 BThomeConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for BThome Bluetooth."""
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self._discovery_info: BluetoothServiceInfo | None = None
self._discovered_device: DeviceData | None = None
self._discovered_devices: dict[str, Discovery] = {}
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfo
) -> FlowResult:
"""Handle the bluetooth discovery step."""
await self.async_set_unique_id(discovery_info.address)
self._abort_if_unique_id_configured()
device = DeviceData()
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._discovery_info = discovery_info
self._discovered_device = device
if device.encryption_scheme == EncryptionScheme.BTHOME_BINDKEY:
return await self.async_step_get_encryption_key()
return await self.async_step_bluetooth_confirm()
async def async_step_get_encryption_key(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Enter a bindkey for an encrypted BThome device."""
assert self._discovery_info
assert self._discovered_device
errors = {}
if user_input is not None:
bindkey = user_input["bindkey"]
if len(bindkey) != 32:
errors["bindkey"] = "expected_32_characters"
else:
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
self._discovered_device.supported(self._discovery_info)
if self._discovered_device.bindkey_verified:
return self._async_get_or_create_entry(bindkey)
errors["bindkey"] = "decryption_failed"
return self.async_show_form(
step_id="get_encryption_key",
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."""
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="bluetooth_confirm",
description_placeholders=self.context["title_placeholders"],
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the user step to pick discovered device."""
if user_input is not None:
address = user_input[CONF_ADDRESS]
await self.async_set_unique_id(address, raise_on_progress=False)
self._abort_if_unique_id_configured()
discovery = self._discovered_devices[address]
self.context["title_placeholders"] = {"name": discovery.title}
self._discovery_info = discovery.discovery_info
self._discovered_device = discovery.device
if discovery.device.encryption_scheme == EncryptionScheme.BTHOME_BINDKEY:
return await self.async_step_get_encryption_key()
return self._async_get_or_create_entry()
current_addresses = self._async_current_ids()
for discovery_info in async_discovered_service_info(self.hass, False):
address = discovery_info.address
if address in current_addresses or address in self._discovered_devices:
continue
device = DeviceData()
if device.supported(discovery_info):
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(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 = entry_data["device"]
self._discovered_device = device
self._discovery_info = device.last_service_info
if device.encryption_scheme == EncryptionScheme.BTHOME_BINDKEY:
return await self.async_step_get_encryption_key()
# 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,
)

3
homeassistant/components/bthome/const.py Normal file
View File

@ -0,0 +1,3 @@
"""Constants for the BThome Bluetooth integration."""
DOMAIN = "bthome"

31
homeassistant/components/bthome/device.py Normal file
View File

@ -0,0 +1,31 @@
"""Support for BThome Bluetooth devices."""
from __future__ import annotations
from bthome_ble import DeviceKey, SensorDeviceInfo
from homeassistant.components.bluetooth.passive_update_processor import (
PassiveBluetoothEntityKey,
)
from homeassistant.const import ATTR_MANUFACTURER, ATTR_MODEL, ATTR_NAME
from homeassistant.helpers.entity import DeviceInfo
def device_key_to_bluetooth_entity_key(
device_key: DeviceKey,
) -> PassiveBluetoothEntityKey:
"""Convert a device key to an entity key."""
return PassiveBluetoothEntityKey(device_key.key, device_key.device_id)
def sensor_device_info_to_hass(
sensor_device_info: SensorDeviceInfo,
) -> DeviceInfo:
"""Convert a sensor device info to a sensor device info."""
hass_device_info = DeviceInfo({})
if sensor_device_info.name is not None:
hass_device_info[ATTR_NAME] = sensor_device_info.name
if sensor_device_info.manufacturer is not None:
hass_device_info[ATTR_MANUFACTURER] = sensor_device_info.manufacturer
if sensor_device_info.model is not None:
hass_device_info[ATTR_MODEL] = sensor_device_info.model
return hass_device_info

20
homeassistant/components/bthome/manifest.json Normal file
View File

@ -0,0 +1,20 @@
{
"domain": "bthome",
"name": "BThome",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/bthome",
"bluetooth": [
{
"connectable": false,
"service_data_uuid": "0000181c-0000-1000-8000-00805f9b34fb"
},
{
"connectable": false,
"service_data_uuid": "0000181e-0000-1000-8000-00805f9b34fb"
}
],
"requirements": ["bthome-ble==0.4.0"],
"dependencies": ["bluetooth"],
"codeowners": ["@Ernst79"],
"iot_class": "local_push"
}

201
homeassistant/components/bthome/sensor.py Normal file
View File

@ -0,0 +1,201 @@
"""Support for BThome sensors."""
from __future__ import annotations
from typing import Optional, Union
from bthome_ble import DeviceClass, SensorUpdate, Units
from homeassistant import config_entries
from homeassistant.components.bluetooth.passive_update_processor import (
PassiveBluetoothDataProcessor,
PassiveBluetoothDataUpdate,
PassiveBluetoothProcessorCoordinator,
PassiveBluetoothProcessorEntity,
)
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
CONCENTRATION_PARTS_PER_MILLION,
ELECTRIC_POTENTIAL_VOLT,
ENERGY_KILO_WATT_HOUR,
LIGHT_LUX,
MASS_KILOGRAMS,
PERCENTAGE,
POWER_WATT,
PRESSURE_MBAR,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .device import device_key_to_bluetooth_entity_key, sensor_device_info_to_hass
SENSOR_DESCRIPTIONS = {
(DeviceClass.TEMPERATURE, Units.TEMP_CELSIUS): SensorEntityDescription(
key=f"{DeviceClass.TEMPERATURE}_{Units.TEMP_CELSIUS}",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=TEMP_CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
),
(DeviceClass.HUMIDITY, Units.PERCENTAGE): SensorEntityDescription(
key=f"{DeviceClass.HUMIDITY}_{Units.PERCENTAGE}",
device_class=SensorDeviceClass.HUMIDITY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
(DeviceClass.ILLUMINANCE, Units.LIGHT_LUX): SensorEntityDescription(
key=f"{DeviceClass.ILLUMINANCE}_{Units.LIGHT_LUX}",
device_class=SensorDeviceClass.ILLUMINANCE,
native_unit_of_measurement=LIGHT_LUX,
state_class=SensorStateClass.MEASUREMENT,
),
(DeviceClass.PRESSURE, Units.PRESSURE_MBAR): SensorEntityDescription(
key=f"{DeviceClass.PRESSURE}_{Units.PRESSURE_MBAR}",
device_class=SensorDeviceClass.PRESSURE,
native_unit_of_measurement=PRESSURE_MBAR,
state_class=SensorStateClass.MEASUREMENT,
),
(DeviceClass.BATTERY, Units.PERCENTAGE): SensorEntityDescription(
key=f"{DeviceClass.BATTERY}_{Units.PERCENTAGE}",
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
),
(DeviceClass.VOLTAGE, Units.ELECTRIC_POTENTIAL_VOLT): SensorEntityDescription(
key=str(Units.ELECTRIC_POTENTIAL_VOLT),
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT,
state_class=SensorStateClass.MEASUREMENT,
),
(DeviceClass.ENERGY, Units.ENERGY_KILO_WATT_HOUR): SensorEntityDescription(
key=str(Units.ENERGY_KILO_WATT_HOUR),
device_class=SensorDeviceClass.ENERGY,
native_unit_of_measurement=ENERGY_KILO_WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
),
(DeviceClass.POWER, Units.POWER_WATT): SensorEntityDescription(
key=str(Units.POWER_WATT),
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=POWER_WATT,
state_class=SensorStateClass.MEASUREMENT,
),
(
DeviceClass.PM10,
Units.CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
): SensorEntityDescription(
key=f"{DeviceClass.PM10}_{Units.CONCENTRATION_MICROGRAMS_PER_CUBIC_METER}",
device_class=SensorDeviceClass.PM10,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT,
),
(
DeviceClass.PM25,
Units.CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
): SensorEntityDescription(
key=f"{DeviceClass.PM25}_{Units.CONCENTRATION_MICROGRAMS_PER_CUBIC_METER}",
device_class=SensorDeviceClass.PM25,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT,
),
(DeviceClass.CO2, Units.CONCENTRATION_PARTS_PER_MILLION,): SensorEntityDescription(
key=f"{DeviceClass.CO2}_{Units.CONCENTRATION_PARTS_PER_MILLION}",
device_class=SensorDeviceClass.CO2,
native_unit_of_measurement=CONCENTRATION_PARTS_PER_MILLION,
state_class=SensorStateClass.MEASUREMENT,
),
(
DeviceClass.VOLATILE_ORGANIC_COMPOUNDS,
Units.CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
): SensorEntityDescription(
key=f"{DeviceClass.VOLATILE_ORGANIC_COMPOUNDS}_{Units.CONCENTRATION_MICROGRAMS_PER_CUBIC_METER}",
device_class=SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS,
native_unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
state_class=SensorStateClass.MEASUREMENT,
),
(
DeviceClass.SIGNAL_STRENGTH,
Units.SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
): SensorEntityDescription(
key=f"{DeviceClass.SIGNAL_STRENGTH}_{Units.SIGNAL_STRENGTH_DECIBELS_MILLIWATT}",
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
# Used for e.g. weight sensor
(None, Units.MASS_KILOGRAMS): SensorEntityDescription(
key=str(Units.MASS_KILOGRAMS),
device_class=None,
native_unit_of_measurement=MASS_KILOGRAMS,
state_class=SensorStateClass.MEASUREMENT,
),
}
def sensor_update_to_bluetooth_data_update(
sensor_update: SensorUpdate,
) -> PassiveBluetoothDataUpdate:
"""Convert a sensor update to a bluetooth data update."""
return PassiveBluetoothDataUpdate(
devices={
device_id: sensor_device_info_to_hass(device_info)
for device_id, device_info in sensor_update.devices.items()
},
entity_descriptions={
device_key_to_bluetooth_entity_key(device_key): SENSOR_DESCRIPTIONS[
(description.device_class, description.native_unit_of_measurement)
]
for device_key, description in sensor_update.entity_descriptions.items()
if description.native_unit_of_measurement
},
entity_data={
device_key_to_bluetooth_entity_key(device_key): sensor_values.native_value
for device_key, sensor_values in sensor_update.entity_values.items()
},
entity_names={
device_key_to_bluetooth_entity_key(device_key): sensor_values.name
for device_key, sensor_values in sensor_update.entity_values.items()
},
)
async def async_setup_entry(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the BThome BLE sensors."""
coordinator: PassiveBluetoothProcessorCoordinator = hass.data[DOMAIN][
entry.entry_id
]
processor = PassiveBluetoothDataProcessor(sensor_update_to_bluetooth_data_update)
entry.async_on_unload(
processor.async_add_entities_listener(
BThomeBluetoothSensorEntity, async_add_entities
)
)
entry.async_on_unload(coordinator.async_register_processor(processor))
class BThomeBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[Optional[Union[float, int]]]
],
SensorEntity,
):
"""Representation of a BThome BLE sensor."""
@property
def native_value(self) -> int | float | None:
"""Return the native value."""
return self.processor.entity_data.get(self.entity_key)

32
homeassistant/components/bthome/strings.json Normal file
View File

@ -0,0 +1,32 @@
{
"config": {
"flow_title": "[%key:component::bluetooth::config::flow_title%]",
"step": {
"user": {
"description": "[%key:component::bluetooth::config::step::user::description%]",
"data": {
"address": "[%key:component::bluetooth::config::step::user::data::address%]"
}
},
"bluetooth_confirm": {
"description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]"
},
"get_encryption_key": {
"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"
}
}
},
"error": {
"decryption_failed": "The provided bindkey did not work, sensor data could not be decrypted. Please check it and try again.",
"expected_32_characters": "Expected a 32 character hexadecimal bindkey."
},
"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%]"
}
}
}

34
homeassistant/components/bthome/translations/en.json Normal file
View File

@ -0,0 +1,34 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"already_in_progress": "Configuration flow is already in progress",
"decryption_failed": "The provided bindkey did not work, sensor data could not be decrypted. Please check it and try again.",
"expected_32_characters": "Expected a 32 character hexadecimal bindkey.",
"no_devices_found": "No devices found on the network",
"reauth_successful": "Re-authentication was successful"
},
"error": {
"decryption_failed": "The provided bindkey did not work, sensor data could not be decrypted. Please check it and try again.",
"expected_32_characters": "Expected a 32 character hexadecimal bindkey."
},
"flow_title": "{name}",
"step": {
"bluetooth_confirm": {
"description": "Do you want to setup {name}?"
},
"get_encryption_key": {
"data": {
"bindkey": "Bindkey"
},
"description": "The sensor data broadcast by the sensor is encrypted. In order to decrypt it we need a 32 character hexadecimal bindkey."
},
"user": {
"data": {
"address": "Device"
},
"description": "Choose a device to setup"
}
}
}
}

10
homeassistant/generated/bluetooth.py
View File

@ -7,6 +7,16 @@ from __future__ import annotations
# fmt: off
BLUETOOTH: list[dict[str, bool | str | int | list[int]]] = [
{
"domain": "bthome",
"connectable": False,
"service_data_uuid": "0000181c-0000-1000-8000-00805f9b34fb"
},
{
"domain": "bthome",
"connectable": False,
"service_data_uuid": "0000181e-0000-1000-8000-00805f9b34fb"
},
{
"domain": "fjaraskupan",
"manufacturer_id": 20296,

1
homeassistant/generated/config_flows.py
View File

@ -57,6 +57,7 @@ FLOWS = {
"brother",
"brunt",
"bsblan",
"bthome",
"buienradar",
"canary",
"cast",

3
requirements_all.txt
View File

@ -457,6 +457,9 @@ bsblan==0.5.0
# homeassistant.components.bluetooth_tracker
bt_proximity==0.2.1
# homeassistant.components.bthome
bthome-ble==0.4.0
# homeassistant.components.bt_home_hub_5
bthomehub5-devicelist==0.1.1

3
requirements_test_all.txt
View File

@ -358,6 +358,9 @@ brunt==1.2.0
# homeassistant.components.bsblan
bsblan==0.5.0
# homeassistant.components.bthome
bthome-ble==0.4.0
# homeassistant.components.buienradar
buienradar==1.0.5

124
tests/components/bthome/__init__.py Normal file
View File

@ -0,0 +1,124 @@
"""Tests for the BThome integration."""
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
TEMP_HUMI_SERVICE_INFO = BluetoothServiceInfoBleak(
name="ATC 8D18B2",
address="A4:C1:38:8D:18:B2",
device=BLEDevice("A4:C1:38:8D:18:B2", None),
rssi=-63,
manufacturer_data={},
service_data={
"0000181c-0000-1000-8000-00805f9b34fb": b"#\x02\xca\t\x03\x03\xbf\x13"
},
service_uuids=["0000181c-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
time=0,
connectable=False,
)
TEMP_HUMI_ENCRYPTED_SERVICE_INFO = BluetoothServiceInfoBleak(
name="TEST DEVICE 8F80A5",
address="54:48:E6:8F:80:A5",
device=BLEDevice("54:48:E6:8F:80:A5", None),
rssi=-63,
manufacturer_data={},
service_data={
"0000181e-0000-1000-8000-00805f9b34fb": b'\xfb\xa45\xe4\xd3\xc3\x12\xfb\x00\x11"3W\xd9\n\x99'
},
service_uuids=["0000181e-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
time=0,
connectable=False,
)
PM_SERVICE_INFO = BluetoothServiceInfoBleak(
name="TEST DEVICE 8F80A5",
address="54:48:E6:8F:80:A5",
device=BLEDevice("54:48:E6:8F:80:A5", None),
rssi=-63,
manufacturer_data={},
service_data={
"0000181c-0000-1000-8000-00805f9b34fb": b"\x03\r\x12\x0c\x03\x0e\x02\x1c"
},
service_uuids=["0000181c-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
time=0,
connectable=False,
)
INVALID_PAYLOAD = BluetoothServiceInfoBleak(
name="ATC 565384",
address="A4:C1:38:56:53:84",
device=BLEDevice("A4:C1:38:56:53:84", None),
rssi=-56,
manufacturer_data={},
service_data={
"0000181c-0000-1000-8000-00805f9b34fb": b"0X[\x05\x02\x84\x53\x568\xc1\xa4\x08",
},
service_uuids=["0000181c-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
time=0,
connectable=False,
)
NOT_BTHOME_SERVICE_INFO = BluetoothServiceInfoBleak(
name="Not it",
address="00:00:00:00:00:00",
device=BLEDevice("00:00:00:00:00:00", None),
rssi=-63,
manufacturer_data={3234: b"\x00\x01"},
service_data={},
service_uuids=[],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
time=0,
connectable=False,
)
def make_advertisement(address: str, payload: bytes) -> BluetoothServiceInfoBleak:
"""Make a dummy advertisement."""
return BluetoothServiceInfoBleak(
name="Test Device",
address=address,
device=BLEDevice(address, None),
rssi=-56,
manufacturer_data={},
service_data={
"0000181c-0000-1000-8000-00805f9b34fb": payload,
},
service_uuids=["0000181c-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Test Device"),
time=0,
connectable=False,
)
def make_encrypted_advertisement(
address: str, payload: bytes
) -> BluetoothServiceInfoBleak:
"""Make a dummy encrypted advertisement."""
return BluetoothServiceInfoBleak(
name="ATC 8F80A5",
address=address,
device=BLEDevice(address, None),
rssi=-56,
manufacturer_data={},
service_data={
"0000181e-0000-1000-8000-00805f9b34fb": payload,
},
service_uuids=["0000181e-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="ATC 8F80A5"),
time=0,
connectable=False,
)

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

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

560
tests/components/bthome/test_config_flow.py Normal file
View File

@ -0,0 +1,560 @@
"""Test the BThome config flow."""
from unittest.mock import patch
from bthome_ble import BThomeBluetoothDeviceData as DeviceData
from homeassistant import config_entries
from homeassistant.components.bluetooth import BluetoothChange
from homeassistant.components.bthome.const import DOMAIN
from homeassistant.data_entry_flow import FlowResultType
from . import (
NOT_BTHOME_SERVICE_INFO,
PM_SERVICE_INFO,
TEMP_HUMI_ENCRYPTED_SERVICE_INFO,
TEMP_HUMI_SERVICE_INFO,
)
from tests.common import MockConfigEntry
async def test_async_step_bluetooth_valid_device(hass):
"""Test discovery via bluetooth with a valid device."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=TEMP_HUMI_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "bluetooth_confirm"
with patch("homeassistant.components.bthome.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"] == "ATC 18B2"
assert result2["data"] == {}
assert result2["result"].unique_id == "A4:C1:38:8D:18:B2"
async def test_async_step_bluetooth_during_onboarding(hass):
"""Test discovery via bluetooth during onboarding."""
with patch(
"homeassistant.components.bthome.async_setup_entry", return_value=True
) as mock_setup_entry, patch(
"homeassistant.components.onboarding.async_is_onboarded",
return_value=False,
) as mock_onboarding:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=TEMP_HUMI_SERVICE_INFO,
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "ATC 18B2"
assert result["data"] == {}
assert result["result"].unique_id == "A4:C1:38:8D:18:B2"
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_onboarding.mock_calls) == 1
async def test_async_step_bluetooth_valid_device_with_encryption(hass):
"""Test discovery via bluetooth with a valid device, with encryption."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=TEMP_HUMI_ENCRYPTED_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "get_encryption_key"
with patch("homeassistant.components.bthome.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "231d39c1d7cc1ab1aee224cd096db932"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "TEST DEVICE 80A5"
assert result2["data"] == {"bindkey": "231d39c1d7cc1ab1aee224cd096db932"}
assert result2["result"].unique_id == "54:48:E6:8F:80:A5"
async def test_async_step_bluetooth_valid_device_encryption_wrong_key(hass):
"""Test discovery via bluetooth with a valid device, with encryption and invalid key."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=TEMP_HUMI_ENCRYPTED_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "get_encryption_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"
assert result2["errors"]["bindkey"] == "decryption_failed"
# Test can finish flow
with patch("homeassistant.components.bthome.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "231d39c1d7cc1ab1aee224cd096db932"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "TEST DEVICE 80A5"
assert result2["data"] == {"bindkey": "231d39c1d7cc1ab1aee224cd096db932"}
assert result2["result"].unique_id == "54:48:E6:8F:80:A5"
async def test_async_step_bluetooth_valid_device_encryption_wrong_key_length(hass):
"""Test discovery via bluetooth with a valid device, with encryption and wrong key length."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=TEMP_HUMI_ENCRYPTED_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "get_encryption_key"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "aa"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "get_encryption_key"
assert result2["errors"]["bindkey"] == "expected_32_characters"
# Test can finish flow
with patch("homeassistant.components.bthome.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "231d39c1d7cc1ab1aee224cd096db932"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "TEST DEVICE 80A5"
assert result2["data"] == {"bindkey": "231d39c1d7cc1ab1aee224cd096db932"}
assert result2["result"].unique_id == "54:48:E6:8F:80:A5"
async def test_async_step_bluetooth_not_supported(hass):
"""Test discovery via bluetooth not supported."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=NOT_BTHOME_SERVICE_INFO,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "not_supported"
async def test_async_step_user_no_devices_found(hass):
"""Test setup from service info cache with no devices found."""
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_no_devices_found_2(hass):
"""
Test setup from service info cache with no devices found.
This variant tests with a non-BThome device known to us.
"""
with patch(
"homeassistant.components.xiaomi_ble.config_flow.async_discovered_service_info",
return_value=[NOT_BTHOME_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(
"homeassistant.components.bthome.config_flow.async_discovered_service_info",
return_value=[PM_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"
with patch("homeassistant.components.bthome.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "54:48:E6:8F:80:A5"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "TEST DEVICE 80A5"
assert result2["data"] == {}
assert result2["result"].unique_id == "54:48:E6:8F:80:A5"
async def test_async_step_user_with_found_devices_encryption(hass):
"""Test setup from service info cache with devices found, with encryption."""
with patch(
"homeassistant.components.bthome.config_flow.async_discovered_service_info",
return_value=[TEMP_HUMI_ENCRYPTED_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:48:E6:8F:80:A5"},
)
assert result1["type"] == FlowResultType.FORM
assert result1["step_id"] == "get_encryption_key"
with patch("homeassistant.components.bthome.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "231d39c1d7cc1ab1aee224cd096db932"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "TEST DEVICE 80A5"
assert result2["data"] == {"bindkey": "231d39c1d7cc1ab1aee224cd096db932"}
assert result2["result"].unique_id == "54:48:E6:8F:80:A5"
async def test_async_step_user_with_found_devices_encryption_wrong_key(hass):
"""Test setup from service info cache with devices found, with encryption and wrong key."""
# Get a list of devices
with patch(
"homeassistant.components.bthome.config_flow.async_discovered_service_info",
return_value=[TEMP_HUMI_ENCRYPTED_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:48:E6:8F:80:A5"},
)
assert result1["type"] == FlowResultType.FORM
assert result1["step_id"] == "get_encryption_key"
# 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"
assert result2["errors"]["bindkey"] == "decryption_failed"
# Check can still finish flow
with patch("homeassistant.components.bthome.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "231d39c1d7cc1ab1aee224cd096db932"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "TEST DEVICE 80A5"
assert result2["data"] == {"bindkey": "231d39c1d7cc1ab1aee224cd096db932"}
assert result2["result"].unique_id == "54:48:E6:8F:80:A5"
async def test_async_step_user_with_found_devices_encryption_wrong_key_length(hass):
"""Test setup from service info cache with devices found, with encryption and wrong key length."""
# Get a list of devices
with patch(
"homeassistant.components.bthome.config_flow.async_discovered_service_info",
return_value=[TEMP_HUMI_ENCRYPTED_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:48:E6:8F:80:A5"},
)
assert result1["type"] == FlowResultType.FORM
assert result1["step_id"] == "get_encryption_key"
# Try an incorrect key
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "aa"},
)
assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "get_encryption_key"
assert result2["errors"]["bindkey"] == "expected_32_characters"
# Check can still finish flow
with patch("homeassistant.components.bthome.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "231d39c1d7cc1ab1aee224cd096db932"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "TEST DEVICE 80A5"
assert result2["data"] == {"bindkey": "231d39c1d7cc1ab1aee224cd096db932"}
assert result2["result"].unique_id == "54:48:E6:8F:80:A5"
async def test_async_step_user_device_added_between_steps(hass):
"""Test the device gets added via another flow between steps."""
with patch(
"homeassistant.components.bthome.config_flow.async_discovered_service_info",
return_value=[TEMP_HUMI_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"
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="A4:C1:38:8D:18:B2",
)
entry.add_to_hass(hass)
with patch("homeassistant.components.bthome.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "A4:C1:38:8D:18:B2"},
)
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "already_configured"
async def test_async_step_user_with_found_devices_already_setup(hass):
"""Test setup from service info cache with devices found."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="A4:C1:38:8D:18:B2",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.bthome.config_flow.async_discovered_service_info",
return_value=[TEMP_HUMI_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_bluetooth_devices_already_setup(hass):
"""Test we can't start a flow if there is already a config entry."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="54:48:E6:8F:80:A5",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=PM_SERVICE_INFO,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_async_step_bluetooth_already_in_progress(hass):
"""Test we can't start a flow for the same device twice."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=PM_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "bluetooth_confirm"
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=PM_SERVICE_INFO,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_in_progress"
async def test_async_step_user_takes_precedence_over_discovery(hass):
"""Test manual setup takes precedence over discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=PM_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "bluetooth_confirm"
with patch(
"homeassistant.components.bthome.config_flow.async_discovered_service_info",
return_value=[PM_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.FORM
with patch("homeassistant.components.bthome.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "54:48:E6:8F:80:A5"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "TEST DEVICE 80A5"
assert result2["data"] == {}
assert result2["result"].unique_id == "54:48:E6:8F:80:A5"
# Verify the original one was aborted
assert not hass.config_entries.flow.async_progress(DOMAIN)
async def test_async_step_reauth(hass):
"""Test reauth with a key."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="54:48:E6:8F:80:A5",
)
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
saved_callback(TEMP_HUMI_ENCRYPTED_SERVICE_INFO, 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"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "231d39c1d7cc1ab1aee224cd096db932"},
)
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "reauth_successful"
async def test_async_step_reauth_wrong_key(hass):
"""Test reauth with a bad key, and that we can recover."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="54:48:E6:8F:80:A5",
)
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
saved_callback(TEMP_HUMI_ENCRYPTED_SERVICE_INFO, 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"
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"
assert result2["errors"]["bindkey"] == "decryption_failed"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"bindkey": "231d39c1d7cc1ab1aee224cd096db932"},
)
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:48:E6:8F:80:A5",
)
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,
},
data=entry.data | {"device": device},
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reauth_successful"

291
tests/components/bthome/test_sensor.py Normal file
View File

@ -0,0 +1,291 @@
"""Test the BThome sensors."""
from unittest.mock import patch
import pytest
from homeassistant.components.bluetooth import BluetoothChange
from homeassistant.components.bthome.const import DOMAIN
from homeassistant.components.sensor import ATTR_STATE_CLASS
from homeassistant.const import ATTR_FRIENDLY_NAME, ATTR_UNIT_OF_MEASUREMENT
from . import make_advertisement, make_encrypted_advertisement
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
"mac_address, advertisement, bind_key, result",
[
(
"A4:C1:38:8D:18:B2",
make_advertisement(
"A4:C1:38:8D:18:B2",
b"#\x02\xca\t\x03\x03\xbf\x13",
),
None,
[
{
"sensor_entity": "sensor.test_device_18b2_temperature",
"friendly_name": "Test Device 18B2 Temperature",
"unit_of_measurement": "°C",
"state_class": "measurement",
"expected_state": "25.06",
},
{
"sensor_entity": "sensor.test_device_18b2_humidity",
"friendly_name": "Test Device 18B2 Humidity",
"unit_of_measurement": "%",
"state_class": "measurement",
"expected_state": "50.55",
},
],
),
(
"A4:C1:38:8D:18:B2",
make_advertisement(
"A4:C1:38:8D:18:B2",
b"\x02\x00\xa8#\x02]\t\x03\x03\xb7\x18\x02\x01]",
),
None,
[
{
"sensor_entity": "sensor.test_device_18b2_temperature",
"friendly_name": "Test Device 18B2 Temperature",
"unit_of_measurement": "°C",
"state_class": "measurement",
"expected_state": "23.97",
},
{
"sensor_entity": "sensor.test_device_18b2_humidity",
"friendly_name": "Test Device 18B2 Humidity",
"unit_of_measurement": "%",
"state_class": "measurement",
"expected_state": "63.27",
},
{
"sensor_entity": "sensor.test_device_18b2_battery",
"friendly_name": "Test Device 18B2 Battery",
"unit_of_measurement": "%",
"state_class": "measurement",
"expected_state": "93",
},
],
),
(
"A4:C1:38:8D:18:B2",
make_advertisement(
"A4:C1:38:8D:18:B2",
b"\x02\x00\x0c\x04\x04\x13\x8a\x01",
),
None,
[
{
"sensor_entity": "sensor.test_device_18b2_pressure",
"friendly_name": "Test Device 18B2 Pressure",
"unit_of_measurement": "mbar",
"state_class": "measurement",
"expected_state": "1008.83",
},
],
),
(
"AA:BB:CC:DD:EE:FF",
make_advertisement(
"AA:BB:CC:DD:EE:FF",
b"\x04\x05\x13\x8a\x14",
),
None,
[
{
"sensor_entity": "sensor.test_device_eeff_illuminance",
"friendly_name": "Test Device EEFF Illuminance",
"unit_of_measurement": "lx",
"state_class": "measurement",
"expected_state": "13460.67",
},
],
),
(
"A4:C1:38:8D:18:B2",
make_advertisement(
"A4:C1:38:8D:18:B2",
b"\x04\n\x13\x8a\x14",
),
None,
[
{
"sensor_entity": "sensor.test_device_18b2_energy",
"friendly_name": "Test Device 18B2 Energy",
"unit_of_measurement": "kWh",
"state_class": "total_increasing",
"expected_state": "1346.067",
},
],
),
(
"A4:C1:38:8D:18:B2",
make_advertisement(
"A4:C1:38:8D:18:B2",
b"\x04\x0b\x02\x1b\x00",
),
None,
[
{
"sensor_entity": "sensor.test_device_18b2_power",
"friendly_name": "Test Device 18B2 Power",
"unit_of_measurement": "W",
"state_class": "measurement",
"expected_state": "69.14",
},
],
),
(
"A4:C1:38:8D:18:B2",
make_advertisement(
"A4:C1:38:8D:18:B2",
b"\x03\x0c\x02\x0c",
),
None,
[
{
"sensor_entity": "sensor.test_device_18b2_voltage",
"friendly_name": "Test Device 18B2 Voltage",
"unit_of_measurement": "V",
"state_class": "measurement",
"expected_state": "3.074",
},
],
),
(
"A4:C1:38:8D:18:B2",
make_advertisement(
"A4:C1:38:8D:18:B2",
b"\x03\r\x12\x0c\x03\x0e\x02\x1c",
),
None,
[
{
"sensor_entity": "sensor.test_device_18b2_pm10",
"friendly_name": "Test Device 18B2 Pm10",
"unit_of_measurement": "µg/m³",
"state_class": "measurement",
"expected_state": "7170",
},
{
"sensor_entity": "sensor.test_device_18b2_pm25",
"friendly_name": "Test Device 18B2 Pm25",
"unit_of_measurement": "µg/m³",
"state_class": "measurement",
"expected_state": "3090",
},
],
),
(
"A4:C1:38:8D:18:B2",
make_advertisement(
"A4:C1:38:8D:18:B2",
b"\x03\x12\xe2\x04",
),
None,
[
{
"sensor_entity": "sensor.test_device_18b2_carbon_dioxide",
"friendly_name": "Test Device 18B2 Carbon Dioxide",
"unit_of_measurement": "ppm",
"state_class": "measurement",
"expected_state": "1250",
},
],
),
(
"A4:C1:38:8D:18:B2",
make_advertisement(
"A4:C1:38:8D:18:B2",
b"\x03\x133\x01",
),
None,
[
{
"sensor_entity": "sensor.test_device_18b2_volatile_organic_compounds",
"friendly_name": "Test Device 18B2 Volatile Organic Compounds",
"unit_of_measurement": "µg/m³",
"state_class": "measurement",
"expected_state": "307",
},
],
),
(
"54:48:E6:8F:80:A5",
make_encrypted_advertisement(
"54:48:E6:8F:80:A5",
b'\xfb\xa45\xe4\xd3\xc3\x12\xfb\x00\x11"3W\xd9\n\x99',
),
"231d39c1d7cc1ab1aee224cd096db932",
[
{
"sensor_entity": "sensor.atc_80a5_temperature",
"friendly_name": "ATC 80A5 Temperature",
"unit_of_measurement": "°C",
"state_class": "measurement",
"expected_state": "25.06",
},
{
"sensor_entity": "sensor.atc_80a5_humidity",
"friendly_name": "ATC 80A5 Humidity",
"unit_of_measurement": "%",
"state_class": "measurement",
"expected_state": "50.55",
},
],
),
],
)
async def test_sensors(
hass,
mac_address,
advertisement,
bind_key,
result,
):
"""Test the different measurement sensors."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=mac_address,
data={"bindkey": bind_key},
)
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
saved_callback(
advertisement,
BluetoothChange.ADVERTISEMENT,
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == len(result)
for meas in result:
sensor = hass.states.get(meas["sensor_entity"])
sensor_attr = sensor.attributes
assert sensor.state == meas["expected_state"]
assert sensor_attr[ATTR_FRIENDLY_NAME] == meas["friendly_name"]
assert sensor_attr[ATTR_UNIT_OF_MEASUREMENT] == meas["unit_of_measurement"]
assert sensor_attr[ATTR_STATE_CLASS] == meas["state_class"]
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()