starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-27 01:46:11 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add support for sleepy BTHome devices (#92991)

Browse Source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Ernst Klamer
2023-05-21 09:01:09 -05:00
committed by GitHub
co-authored by J. Nick Koston
parent 5821a563c7
commit 40c74ad5b7
10 changed files with 345 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/components/bthome/test_binary_sensor.py
+127 -4
View File
@@ -1,16 +1,31 @@
"""Test BTHome binary sensors."""
from datetime import timedelta
import logging
import time
from unittest.mock import patch
import pytest
from homeassistant.components.bluetooth import (
FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
)
from homeassistant.components.bthome.const import DOMAIN
from homeassistant.const import ATTR_FRIENDLY_NAME, STATE_OFF, STATE_ON
from homeassistant.const import (
ATTR_FRIENDLY_NAME,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from . import make_bthome_v1_adv, make_bthome_v2_adv
from tests.common import MockConfigEntry
from tests.components.bluetooth import inject_bluetooth_service_info
from tests.common import MockConfigEntry, async_fire_time_changed
from tests.components.bluetooth import (
inject_bluetooth_service_info,
patch_all_discovered_devices,
)
_LOGGER = logging.getLogger(__name__)
@@ -123,7 +138,7 @@ async def test_v1_binary_sensors(
"A4:C1:38:8D:18:B2",
make_bthome_v2_adv(
"A4:C1:38:8D:18:B2",
b"\x40\x11\x00",
b"\x44\x11\x00",
),
None,
[
@@ -185,3 +200,111 @@ async def test_v2_binary_sensors(
assert binary_sensor_attr[ATTR_FRIENDLY_NAME] == meas["friendly_name"]
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_unavailable(hass: HomeAssistant) -> None:
"""Test normal device goes to unavailable after 60 minutes."""
start_monotonic = time.monotonic()
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="A4:C1:38:8D:18:B2",
data={},
)
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,
make_bthome_v2_adv(
"A4:C1:38:8D:18:B2",
b"\x40\x11\x01",
),
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
opening_sensor = hass.states.get("binary_sensor.test_device_18b2_opening")
assert opening_sensor.state == STATE_ON
# Fastforward time without BLE advertisements
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now,
), patch_all_discovered_devices([]):
async_fire_time_changed(
hass,
dt_util.utcnow()
+ timedelta(seconds=FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1),
)
await hass.async_block_till_done()
opening_sensor = hass.states.get("binary_sensor.test_device_18b2_opening")
# Normal devices should go to unavailable
assert opening_sensor.state == STATE_UNAVAILABLE
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_sleepy_device(hass: HomeAssistant) -> None:
"""Test sleepy device does not go to unavailable after 60 minutes."""
start_monotonic = time.monotonic()
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="A4:C1:38:8D:18:B2",
data={},
)
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,
make_bthome_v2_adv(
"A4:C1:38:8D:18:B2",
b"\x44\x11\x01",
),
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
opening_sensor = hass.states.get("binary_sensor.test_device_18b2_opening")
assert opening_sensor.state == STATE_ON
# Fastforward time without BLE advertisements
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now,
), patch_all_discovered_devices([]):
async_fire_time_changed(
hass,
dt_util.utcnow()
+ timedelta(seconds=FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1),
)
await hass.async_block_till_done()
opening_sensor = hass.states.get("binary_sensor.test_device_18b2_opening")
# Sleepy devices should keep their state over time
assert opening_sensor.state == STATE_ON
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
tests/components/bthome/test_sensor.py
+126 -6
View File
@@ -1,17 +1,31 @@
"""Test the BTHome sensors."""
from datetime import timedelta
import logging
import time
from unittest.mock import patch
import pytest
from homeassistant.components.bluetooth import (
FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
)
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 homeassistant.const import (
ATTR_FRIENDLY_NAME,
ATTR_UNIT_OF_MEASUREMENT,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from . import make_bthome_v1_adv, make_bthome_v2_adv, make_encrypted_bthome_v1_adv
from tests.common import MockConfigEntry
from tests.components.bluetooth import inject_bluetooth_service_info
from tests.common import MockConfigEntry, async_fire_time_changed
from tests.components.bluetooth import (
inject_bluetooth_service_info,
patch_all_discovered_devices,
)
_LOGGER = logging.getLogger(__name__)
@@ -518,7 +532,7 @@ async def test_v1_sensors(
"A4:C1:38:8D:18:B2",
make_bthome_v2_adv(
"A4:C1:38:8D:18:B2",
b"\x40\x09\x60",
b"\x44\x09\x60",
),
None,
[
@@ -1008,9 +1022,7 @@ async def test_v2_sensors(
assert len(hass.states.async_all()) == len(result)
for meas in result:
_LOGGER.error(meas)
sensor = hass.states.get(meas["sensor_entity"])
_LOGGER.error(hass.states)
sensor_attr = sensor.attributes
assert sensor.state == meas["expected_state"]
assert sensor_attr[ATTR_FRIENDLY_NAME] == meas["friendly_name"]
@@ -1020,3 +1032,111 @@ async def test_v2_sensors(
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()
async def test_unavailable(hass: HomeAssistant) -> None:
"""Test normal device goes to unavailable after 60 minutes."""
start_monotonic = time.monotonic()
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="A4:C1:38:8D:18:B2",
data={},
)
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,
make_bthome_v2_adv(
"A4:C1:38:8D:18:B2",
b"\x40\x04\x13\x8a\x01",
),
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
pressure_sensor = hass.states.get("sensor.test_device_18b2_pressure")
assert pressure_sensor.state == "1008.83"
# Fastforward time without BLE advertisements
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now,
), patch_all_discovered_devices([]):
async_fire_time_changed(
hass,
dt_util.utcnow()
+ timedelta(seconds=FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1),
)
await hass.async_block_till_done()
pressure_sensor = hass.states.get("sensor.test_device_18b2_pressure")
# Normal devices should go to unavailable
assert pressure_sensor.state == STATE_UNAVAILABLE
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_sleepy_device(hass: HomeAssistant) -> None:
"""Test sleepy device does not go to unavailable after 60 minutes."""
start_monotonic = time.monotonic()
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="A4:C1:38:8D:18:B2",
data={},
)
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,
make_bthome_v2_adv(
"A4:C1:38:8D:18:B2",
b"\x44\x04\x13\x8a\x01",
),
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
pressure_sensor = hass.states.get("sensor.test_device_18b2_pressure")
assert pressure_sensor.state == "1008.83"
# Fastforward time without BLE advertisements
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now,
), patch_all_discovered_devices([]):
async_fire_time_changed(
hass,
dt_util.utcnow()
+ timedelta(seconds=FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1),
)
await hass.async_block_till_done()
pressure_sensor = hass.states.get("sensor.test_device_18b2_pressure")
# Sleepy devices should keep their state over time
assert pressure_sensor.state == "1008.83"
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()