297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""Test the INKBIRD config flow."""
|
|
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from inkbird_ble import (
|
|
DeviceKey,
|
|
INKBIRDBluetoothDeviceData,
|
|
SensorDescription,
|
|
SensorDeviceInfo,
|
|
SensorUpdate,
|
|
SensorValue,
|
|
Units,
|
|
)
|
|
from inkbird_ble.parser import Model
|
|
from sensor_state_data import SensorDeviceClass
|
|
|
|
from homeassistant.components.inkbird.const import (
|
|
CONF_DEVICE_DATA,
|
|
CONF_DEVICE_TYPE,
|
|
DOMAIN,
|
|
)
|
|
from homeassistant.components.inkbird.coordinator import FALLBACK_POLL_INTERVAL
|
|
from homeassistant.components.sensor import ATTR_STATE_CLASS
|
|
from homeassistant.config_entries import ConfigEntryState
|
|
from homeassistant.const import ATTR_FRIENDLY_NAME, ATTR_UNIT_OF_MEASUREMENT
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.util import dt as dt_util
|
|
|
|
from . import (
|
|
IAM_T1_SERVICE_INFO,
|
|
IBS_P02B_SERVICE_INFO,
|
|
SPS_PASSIVE_SERVICE_INFO,
|
|
SPS_SERVICE_INFO,
|
|
SPS_WITH_CORRUPT_NAME_SERVICE_INFO,
|
|
)
|
|
|
|
from tests.common import MockConfigEntry, async_fire_time_changed
|
|
from tests.components.bluetooth import inject_bluetooth_service_info
|
|
|
|
|
|
def _make_sensor_update(name: str, humidity: float) -> SensorUpdate:
|
|
return SensorUpdate(
|
|
title=None,
|
|
devices={
|
|
None: SensorDeviceInfo(
|
|
name=f"{name} EEFF",
|
|
model=name,
|
|
manufacturer="INKBIRD",
|
|
sw_version=None,
|
|
hw_version=None,
|
|
)
|
|
},
|
|
entity_descriptions={
|
|
DeviceKey(key="humidity", device_id=None): SensorDescription(
|
|
device_key=DeviceKey(key="humidity", device_id=None),
|
|
device_class=SensorDeviceClass.HUMIDITY,
|
|
native_unit_of_measurement=Units.PERCENTAGE,
|
|
),
|
|
},
|
|
entity_values={
|
|
DeviceKey(key="humidity", device_id=None): SensorValue(
|
|
device_key=DeviceKey(key="humidity", device_id=None),
|
|
name="Humidity",
|
|
native_value=humidity,
|
|
),
|
|
},
|
|
)
|
|
|
|
|
|
async def test_sensors(hass: HomeAssistant) -> None:
|
|
"""Test setting up creates the sensors."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
unique_id="61DE521B-F0BF-9F44-64D4-75BBE1738105",
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(hass.states.async_all()) == 0
|
|
inject_bluetooth_service_info(hass, SPS_SERVICE_INFO)
|
|
await hass.async_block_till_done()
|
|
assert len(hass.states.async_all()) == 3
|
|
|
|
temp_sensor = hass.states.get("sensor.ibs_th_8105_battery")
|
|
temp_sensor_attribtes = temp_sensor.attributes
|
|
assert temp_sensor.state == "87"
|
|
assert temp_sensor_attribtes[ATTR_FRIENDLY_NAME] == "IBS-TH 8105 Battery"
|
|
assert temp_sensor_attribtes[ATTR_UNIT_OF_MEASUREMENT] == "%"
|
|
assert temp_sensor_attribtes[ATTR_STATE_CLASS] == "measurement"
|
|
|
|
# Make sure we remember the device type
|
|
# in case the name is corrupted later
|
|
assert entry.data[CONF_DEVICE_TYPE] == "IBS-TH"
|
|
assert await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
|
|
async def test_device_with_corrupt_name(hass: HomeAssistant) -> None:
|
|
"""Test setting up a known device type with a corrupt name."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
unique_id="AA:BB:CC:DD:EE:FF",
|
|
data={CONF_DEVICE_TYPE: "IBS-TH"},
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(hass.states.async_all()) == 0
|
|
inject_bluetooth_service_info(hass, SPS_WITH_CORRUPT_NAME_SERVICE_INFO)
|
|
await hass.async_block_till_done()
|
|
assert len(hass.states.async_all()) == 3
|
|
|
|
temp_sensor = hass.states.get("sensor.ibs_th_eeff_battery")
|
|
temp_sensor_attribtes = temp_sensor.attributes
|
|
assert temp_sensor.state == "87"
|
|
assert temp_sensor_attribtes[ATTR_FRIENDLY_NAME] == "IBS-TH EEFF Battery"
|
|
assert temp_sensor_attribtes[ATTR_UNIT_OF_MEASUREMENT] == "%"
|
|
assert temp_sensor_attribtes[ATTR_STATE_CLASS] == "measurement"
|
|
|
|
assert entry.data[CONF_DEVICE_TYPE] == "IBS-TH"
|
|
assert await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
|
|
async def test_polling_sensor(hass: HomeAssistant) -> None:
|
|
"""Test setting up a device that needs polling."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
unique_id="AA:BB:CC:DD:EE:FF",
|
|
data={CONF_DEVICE_TYPE: "IBS-TH"},
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(hass.states.async_all()) == 0
|
|
|
|
with patch(
|
|
"homeassistant.components.inkbird.coordinator.INKBIRDBluetoothDeviceData.async_poll",
|
|
return_value=_make_sensor_update("IBS-TH", 10.24),
|
|
):
|
|
inject_bluetooth_service_info(hass, SPS_PASSIVE_SERVICE_INFO)
|
|
await hass.async_block_till_done()
|
|
assert len(hass.states.async_all()) == 1
|
|
|
|
temp_sensor = hass.states.get("sensor.ibs_th_eeff_humidity")
|
|
temp_sensor_attribtes = temp_sensor.attributes
|
|
assert temp_sensor.state == "10.24"
|
|
assert temp_sensor_attribtes[ATTR_FRIENDLY_NAME] == "IBS-TH EEFF Humidity"
|
|
assert temp_sensor_attribtes[ATTR_UNIT_OF_MEASUREMENT] == "%"
|
|
assert temp_sensor_attribtes[ATTR_STATE_CLASS] == "measurement"
|
|
|
|
assert entry.data[CONF_DEVICE_TYPE] == "IBS-TH"
|
|
|
|
with patch(
|
|
"homeassistant.components.inkbird.coordinator.INKBIRDBluetoothDeviceData.async_poll",
|
|
return_value=_make_sensor_update("IBS-TH", 20.24),
|
|
):
|
|
async_fire_time_changed(hass, dt_util.utcnow() + FALLBACK_POLL_INTERVAL)
|
|
inject_bluetooth_service_info(hass, SPS_PASSIVE_SERVICE_INFO)
|
|
await hass.async_block_till_done()
|
|
|
|
temp_sensor = hass.states.get("sensor.ibs_th_eeff_humidity")
|
|
temp_sensor_attribtes = temp_sensor.attributes
|
|
assert temp_sensor.state == "20.24"
|
|
|
|
assert await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
|
|
async def test_notify_sensor_no_advertisement(hass: HomeAssistant) -> None:
|
|
"""Test setting up a notify sensor that has no advertisement."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
unique_id="62:00:A1:3C:AE:7B",
|
|
data={CONF_DEVICE_TYPE: "IAM-T1"},
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
assert not await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert entry.state is ConfigEntryState.SETUP_RETRY
|
|
|
|
|
|
async def test_notify_sensor(hass: HomeAssistant) -> None:
|
|
"""Test setting up a notify sensor."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
unique_id="62:00:A1:3C:AE:7B",
|
|
data={CONF_DEVICE_TYPE: "IAM-T1"},
|
|
)
|
|
entry.add_to_hass(hass)
|
|
inject_bluetooth_service_info(hass, IAM_T1_SERVICE_INFO)
|
|
saved_update_callback = None
|
|
saved_device_data_changed_callback = None
|
|
|
|
class MockINKBIRDBluetoothDeviceData(INKBIRDBluetoothDeviceData):
|
|
def __init__(
|
|
self,
|
|
device_type: Model | str | None = None,
|
|
device_data: dict[str, Any] | None = None,
|
|
update_callback: Callable[[SensorUpdate], None] | None = None,
|
|
device_data_changed_callback: Callable[[dict[str, Any]], None]
|
|
| None = None,
|
|
) -> None:
|
|
nonlocal saved_update_callback
|
|
nonlocal saved_device_data_changed_callback
|
|
saved_update_callback = update_callback
|
|
saved_device_data_changed_callback = device_data_changed_callback
|
|
super().__init__(
|
|
device_type=device_type,
|
|
device_data=device_data,
|
|
update_callback=update_callback,
|
|
device_data_changed_callback=device_data_changed_callback,
|
|
)
|
|
|
|
mock_client = MagicMock(start_notify=AsyncMock(), disconnect=AsyncMock())
|
|
with (
|
|
patch(
|
|
"homeassistant.components.inkbird.coordinator.INKBIRDBluetoothDeviceData",
|
|
MockINKBIRDBluetoothDeviceData,
|
|
),
|
|
patch("inkbird_ble.parser.establish_connection", return_value=mock_client),
|
|
):
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert entry.state is ConfigEntryState.LOADED
|
|
assert len(hass.states.async_all()) == 0
|
|
|
|
saved_update_callback(_make_sensor_update("IAM-T1", 10.24))
|
|
|
|
assert len(hass.states.async_all()) == 1
|
|
|
|
temp_sensor = hass.states.get("sensor.iam_t1_eeff_humidity")
|
|
temp_sensor_attribtes = temp_sensor.attributes
|
|
assert temp_sensor.state == "10.24"
|
|
assert temp_sensor_attribtes[ATTR_FRIENDLY_NAME] == "IAM-T1 EEFF Humidity"
|
|
assert temp_sensor_attribtes[ATTR_UNIT_OF_MEASUREMENT] == "%"
|
|
assert temp_sensor_attribtes[ATTR_STATE_CLASS] == "measurement"
|
|
|
|
assert entry.data[CONF_DEVICE_TYPE] == "IAM-T1"
|
|
|
|
saved_device_data_changed_callback({"temp_unit": "F"})
|
|
assert entry.data[CONF_DEVICE_DATA] == {"temp_unit": "F"}
|
|
|
|
saved_device_data_changed_callback({"temp_unit": "C"})
|
|
assert entry.data[CONF_DEVICE_DATA] == {"temp_unit": "C"}
|
|
|
|
saved_device_data_changed_callback({"temp_unit": "C"})
|
|
assert entry.data[CONF_DEVICE_DATA] == {"temp_unit": "C"}
|
|
|
|
|
|
async def test_ibs_p02b_sensors(hass: HomeAssistant) -> None:
|
|
"""Test setting up creates the sensors for an IBS-P02B."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
unique_id="49:24:11:18:00:65",
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(hass.states.async_all()) == 0
|
|
inject_bluetooth_service_info(hass, IBS_P02B_SERVICE_INFO)
|
|
await hass.async_block_till_done()
|
|
assert len(hass.states.async_all()) == 2
|
|
|
|
temp_sensor = hass.states.get("sensor.ibs_p02b_0065_battery")
|
|
temp_sensor_attribtes = temp_sensor.attributes
|
|
assert temp_sensor.state == "95"
|
|
assert temp_sensor_attribtes[ATTR_FRIENDLY_NAME] == "IBS-P02B 0065 Battery"
|
|
assert temp_sensor_attribtes[ATTR_UNIT_OF_MEASUREMENT] == "%"
|
|
assert temp_sensor_attribtes[ATTR_STATE_CLASS] == "measurement"
|
|
|
|
temp_sensor = hass.states.get("sensor.ibs_p02b_0065_temperature")
|
|
temp_sensor_attribtes = temp_sensor.attributes
|
|
assert temp_sensor.state == "36.6"
|
|
assert temp_sensor_attribtes[ATTR_FRIENDLY_NAME] == "IBS-P02B 0065 Temperature"
|
|
assert temp_sensor_attribtes[ATTR_UNIT_OF_MEASUREMENT] == "°C"
|
|
assert temp_sensor_attribtes[ATTR_STATE_CLASS] == "measurement"
|
|
|
|
# Make sure we remember the device type
|
|
# in case the name is corrupted later
|
|
assert entry.data[CONF_DEVICE_TYPE] == "IBS-P02B"
|
|
assert await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|