Compare commits

...

2 Commits

Author SHA1 Message Date
Erik 03e05083a8 Address review comments 2026-05-19 09:58:11 +02:00
Erik e76e923f26 Add new device tracker base entity BaseScannerEntity 2026-05-18 10:19:22 +02:00
5 changed files with 197 additions and 48 deletions
@@ -5,6 +5,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType
from .config_entry import ( # noqa: F401
BaseScannerEntity,
ScannerEntity,
ScannerEntityDescription,
TrackerEntity,
@@ -166,7 +166,11 @@ def _async_register_mac(
class BaseTrackerEntity(Entity):
"""Represent a tracked device."""
"""Represent a tracked device.
Not intended to be directly inherited by integrations. Integrations should
inherit TrackerEntity, BaseScannerEntity or ScannerEntity instead.
"""
_attr_device_info: None = None
_attr_entity_category = EntityCategory.DIAGNOSTIC
@@ -304,6 +308,28 @@ class TrackerEntity(
return attr
class BaseScannerEntity(BaseTrackerEntity):
"""Base class for a tracked device that can be connected or disconnected.
Unlike ScannerEntity, this entity does not make assumptions about MAC
addresses being used to identify the device.
"""
@property
def state(self) -> str | None:
"""Return the state of the device."""
if self.is_connected is None:
return None
if self.is_connected:
return STATE_HOME
return STATE_NOT_HOME
@property
def is_connected(self) -> bool | None:
"""Return true if the device is connected."""
raise NotImplementedError
class ScannerEntityDescription(EntityDescription, frozen_or_thawed=True):
"""A class that describes tracker entities."""
@@ -316,7 +342,7 @@ CACHED_SCANNER_PROPERTIES_WITH_ATTR_ = {
class ScannerEntity(
BaseTrackerEntity, cached_properties=CACHED_SCANNER_PROPERTIES_WITH_ATTR_
BaseScannerEntity, cached_properties=CACHED_SCANNER_PROPERTIES_WITH_ATTR_
):
"""Base class for a tracked device that is on a scanned network."""
@@ -341,18 +367,6 @@ class ScannerEntity(
"""Return hostname of the device."""
return self._attr_hostname
@property
def state(self) -> str:
"""Return the state of the device."""
if self.is_connected:
return STATE_HOME
return STATE_NOT_HOME
@property
def is_connected(self) -> bool:
"""Return true if the device is connected to the network."""
raise NotImplementedError
@property
def unique_id(self) -> str | None:
"""Return unique ID of the entity."""
@@ -2,9 +2,7 @@
from ibeacon_ble import iBeaconAdvertisement
from homeassistant.components.device_tracker import SourceType
from homeassistant.components.device_tracker.config_entry import BaseTrackerEntity
from homeassistant.const import STATE_HOME, STATE_NOT_HOME
from homeassistant.components.device_tracker import BaseScannerEntity, SourceType
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
@@ -46,10 +44,11 @@ async def async_setup_entry(
)
class IBeaconTrackerEntity(IBeaconEntity, BaseTrackerEntity):
class IBeaconTrackerEntity(IBeaconEntity, BaseScannerEntity):
"""An iBeacon Tracker entity."""
_attr_name = None
_attr_source_type: SourceType = SourceType.BLUETOOTH_LE
_attr_translation_key = "device_tracker"
def __init__(
@@ -67,14 +66,9 @@ class IBeaconTrackerEntity(IBeaconEntity, BaseTrackerEntity):
self._active = True
@property
def state(self) -> str:
"""Return the state of the device."""
return STATE_HOME if self._active else STATE_NOT_HOME
@property
def source_type(self) -> SourceType:
"""Return tracker source type."""
return SourceType.BLUETOOTH_LE
def is_connected(self) -> bool:
"""Return true if the device is connected."""
return self._active
@callback
def _async_seen(
@@ -4,10 +4,8 @@ from collections.abc import Mapping
import logging
from homeassistant.components import bluetooth
from homeassistant.components.device_tracker import SourceType
from homeassistant.components.device_tracker.config_entry import BaseTrackerEntity
from homeassistant.components.device_tracker import BaseScannerEntity, SourceType
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_HOME, STATE_NOT_HOME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
@@ -25,11 +23,12 @@ async def async_setup_entry(
async_add_entities([BasePrivateDeviceTracker(config_entry)])
class BasePrivateDeviceTracker(BasePrivateDeviceEntity, BaseTrackerEntity):
class BasePrivateDeviceTracker(BasePrivateDeviceEntity, BaseScannerEntity):
"""A trackable Private Bluetooth Device."""
_attr_should_poll = False
_attr_has_entity_name = True
_attr_source_type: SourceType = SourceType.BLUETOOTH_LE
_attr_translation_key = "device_tracker"
_attr_name = None
@@ -60,11 +59,6 @@ class BasePrivateDeviceTracker(BasePrivateDeviceEntity, BaseTrackerEntity):
self.async_write_ha_state()
@property
def state(self) -> str:
"""Return the state of the device."""
return STATE_HOME if self._last_info else STATE_NOT_HOME
@property
def source_type(self) -> SourceType:
"""Return the source type, eg gps or router, of the device."""
return SourceType.BLUETOOTH_LE
def is_connected(self) -> bool:
"""Return true if the device is connected."""
return bool(self._last_info)
@@ -16,6 +16,7 @@ from homeassistant.components.device_tracker import (
)
from homeassistant.components.device_tracker.config_entry import (
CONNECTED_DEVICE_REGISTERED,
BaseScannerEntity,
BaseTrackerEntity,
ScannerEntity,
TrackerEntity,
@@ -242,6 +243,64 @@ def tracker_entity_fixture(
return entity
class MockBaseScannerEntity(BaseScannerEntity):
"""Test base scanner entity."""
def __init__(
self,
connected: bool | None = False,
unique_id: str | None = None,
) -> None:
"""Initialize entity."""
self._connected = connected
self._unique_id = unique_id
@property
def should_poll(self) -> bool:
"""Return False for the test entity."""
return False
@property
def source_type(self) -> SourceType:
"""Return the source type, eg gps or router, of the device."""
return SourceType.BLUETOOTH_LE
@property
def is_connected(self) -> bool | None:
"""Return true if the device is connected to the network."""
return self._connected
@property
def unique_id(self) -> str | None:
"""Return hostname of the device."""
return self._unique_id
@callback
def set_connected(self, connected: bool | None) -> None:
"""Set connected state."""
self._connected = connected
self.async_write_ha_state()
@pytest.fixture(name="unique_id")
def unique_id_fixture() -> str | None:
"""Return the unique_id of the entity for the test."""
return None
@pytest.fixture(name="base_scanner_entity")
def base_scanner_entity_fixture(
entity_id: str,
unique_id: str | None,
) -> MockBaseScannerEntity:
"""Create a test base scanner entity."""
entity = MockBaseScannerEntity(
unique_id=unique_id,
)
entity.entity_id = entity_id
return entity
class MockScannerEntity(ScannerEntity):
"""Test scanner entity."""
@@ -250,7 +309,7 @@ class MockScannerEntity(ScannerEntity):
ip_address: str | None = None,
mac_address: str | None = None,
hostname: str | None = None,
connected: bool = False,
connected: bool | None = False,
unique_id: str | None = None,
) -> None:
"""Initialize entity."""
@@ -286,7 +345,7 @@ class MockScannerEntity(ScannerEntity):
return self._hostname
@property
def is_connected(self) -> bool:
def is_connected(self) -> bool | None:
"""Return true if the device is connected to the network."""
return self._connected
@@ -296,7 +355,7 @@ class MockScannerEntity(ScannerEntity):
return self._unique_id or self._mac_address
@callback
def set_connected(self, connected: bool) -> None:
def set_connected(self, connected: bool | None) -> None:
"""Set connected state."""
self._connected = connected
self.async_write_ha_state()
@@ -320,12 +379,6 @@ def hostname_fixture() -> str | None:
return None
@pytest.fixture(name="unique_id")
def unique_id_fixture() -> str | None:
"""Return the unique_id of the entity for the test."""
return None
@pytest.fixture(name="scanner_entity")
def scanner_entity_fixture(
entity_id: str,
@@ -345,7 +398,49 @@ def scanner_entity_fixture(
return entity
async def test_load_unload_entry(
async def test_load_unload_entry_base_scanner(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_id: str,
base_scanner_entity: MockBaseScannerEntity,
) -> None:
"""Test loading and unloading a config entry with a device tracker entity."""
config_entry = await create_mock_platform(hass, config_entry, [base_scanner_entity])
assert config_entry.state is ConfigEntryState.LOADED
state = hass.states.get(entity_id)
assert state
assert await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.NOT_LOADED
state = hass.states.get(entity_id)
assert not state
async def test_load_unload_entry_scanner(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_id: str,
scanner_entity: MockScannerEntity,
) -> None:
"""Test loading and unloading a config entry with a device tracker entity."""
config_entry = await create_mock_platform(hass, config_entry, [scanner_entity])
assert config_entry.state is ConfigEntryState.LOADED
state = hass.states.get(entity_id)
assert state
assert await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.NOT_LOADED
state = hass.states.get(entity_id)
assert not state
async def test_load_unload_entry_tracker(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_id: str,
@@ -490,6 +585,38 @@ async def test_tracker_entity_state(
assert state.attributes == expected_attributes
async def test_base_scanner_entity_state(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_id: str,
base_scanner_entity: MockBaseScannerEntity,
) -> None:
"""Test BaseScannerEntity based device tracker."""
config_entry = await create_mock_platform(hass, config_entry, [base_scanner_entity])
assert config_entry.state is ConfigEntryState.LOADED
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.attributes == {
ATTR_SOURCE_TYPE: SourceType.BLUETOOTH_LE,
}
assert entity_state.state == STATE_NOT_HOME
base_scanner_entity.set_connected(True)
await hass.async_block_till_done()
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_HOME
base_scanner_entity.set_connected(None)
await hass.async_block_till_done()
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_UNKNOWN
@pytest.mark.parametrize(
("ip_address", "mac_address", "hostname"),
[("0.0.0.0", "ad:de:ef:be:ed:fe", "test.hostname.org")],
@@ -535,6 +662,13 @@ async def test_scanner_entity_state(
assert entity_state
assert entity_state.state == STATE_HOME
scanner_entity.set_connected(None)
await hass.async_block_till_done()
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_UNKNOWN
def test_tracker_entity() -> None:
"""Test coverage for base TrackerEntity class."""
@@ -570,6 +704,18 @@ def test_tracker_entity() -> None:
assert not test_entity.force_update
def test_base_scanner_entity() -> None:
"""Test coverage for base BaseScannerEntity entity class."""
entity = BaseScannerEntity()
with pytest.raises(NotImplementedError):
entity.source_type # noqa: B018
with pytest.raises(NotImplementedError):
entity.is_connected # noqa: B018
with pytest.raises(NotImplementedError):
entity.state # noqa: B018
assert entity.battery_level is None
def test_scanner_entity() -> None:
"""Test coverage for base ScannerEntity entity class."""
entity = ScannerEntity()