Compare commits

...

1 Commits

Author SHA1 Message Date
Erik 6f2b83411d Add device_tracker platform to kitchen_sink 2026-05-26 12:22:48 +02:00
8 changed files with 451 additions and 30 deletions
@@ -8,8 +8,6 @@ import datetime
from functools import partial
from random import random
import voluptuous as vol
from homeassistant.components.labs import (
EventLabsUpdatedData,
async_is_preview_feature_enabled,
@@ -34,7 +32,7 @@ from homeassistant.const import (
UnitOfTemperature,
UnitOfVolume,
)
from homeassistant.core import HomeAssistant, ServiceCall, ServiceResponse, callback
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.device_registry import DeviceEntry
from homeassistant.helpers.issue_registry import (
@@ -51,9 +49,11 @@ from homeassistant.util.unit_conversion import (
)
from .const import DATA_BACKUP_AGENT_LISTENERS, DOMAIN
from .services import async_setup_services
COMPONENTS_WITH_DEMO_PLATFORM = [
Platform.BUTTON,
Platform.DEVICE_TRACKER,
Platform.FAN,
Platform.EVENT,
Platform.IMAGE,
@@ -69,15 +69,6 @@ COMPONENTS_WITH_DEMO_PLATFORM = [
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
SCHEMA_SERVICE_TEST_SERVICE_1 = vol.Schema(
{
vol.Required("field_1"): vol.Coerce(int),
vol.Required("field_2"): vol.In(["off", "auto", "cool"]),
vol.Optional("field_3"): vol.Coerce(int),
vol.Optional("field_4"): vol.In(["forwards", "reverse"]),
}
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the demo environment."""
@@ -87,24 +78,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
)
)
@callback
def service_handler(call: ServiceCall | None = None) -> ServiceResponse:
"""Do nothing."""
return None
hass.services.async_register(
DOMAIN,
"test_service_1",
service_handler,
SCHEMA_SERVICE_TEST_SERVICE_1,
description_placeholders={
"meep_1": "foo",
"meep_2": "bar",
"meep_3": "beer",
"meep_4": "milk",
"meep_5": "https://example.com",
},
)
async_setup_services(hass)
return True
@@ -0,0 +1,97 @@
"""Demo platform that has a couple of fake device trackers."""
from homeassistant.components.device_tracker import (
BaseScannerEntity,
SourceType,
TrackerEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Everything but the Kitchen Sink config entry."""
async_add_entities(
[
DemoTracker(
unique_id="kitchen_sink_tracker_001",
name="Demo tracker",
latitude=hass.config.latitude,
longitude=hass.config.longitude,
accuracy=10,
),
DemoScanner(
unique_id="kitchen_sink_scanner_001",
name="Demo scanner",
is_connected=True,
),
]
)
class DemoTracker(TrackerEntity):
"""Representation of a demo tracker."""
_attr_should_poll = False
_attr_source_type = SourceType.GPS
def __init__(
self,
*,
unique_id: str,
name: str,
latitude: float | None,
longitude: float | None,
accuracy: float,
) -> None:
"""Initialize the tracker."""
self._attr_unique_id = unique_id
self._attr_name = name
self._attr_latitude = latitude
self._attr_longitude = longitude
self._attr_location_accuracy = accuracy
@callback
def async_set_tracker_location(
self, latitude: float, longitude: float, accuracy: float
) -> None:
"""Update the tracker location."""
self._attr_latitude = latitude
self._attr_longitude = longitude
self._attr_location_accuracy = accuracy
self.async_write_ha_state()
class DemoScanner(BaseScannerEntity):
"""Representation of a demo scanner."""
_attr_should_poll = False
_attr_source_type = SourceType.ROUTER
def __init__(
self,
*,
unique_id: str,
name: str,
is_connected: bool,
) -> None:
"""Initialize the scanner."""
self._attr_unique_id = unique_id
self._attr_name = name
self._is_connected = is_connected
@property
def is_connected(self) -> bool:
"""Return true if the device is connected."""
return self._is_connected
@callback
def async_set_scanner_connected(self, connected: bool) -> None:
"""Update the scanner connected state."""
self._is_connected = connected
self.async_write_ha_state()
@@ -9,6 +9,12 @@
}
},
"services": {
"set_scanner_connected": {
"service": "mdi:lan-connect"
},
"set_tracker_location": {
"service": "mdi:map-marker"
},
"test_service_1": {
"sections": {
"additional_fields": "mdi:test-tube"
@@ -0,0 +1,72 @@
"""Services for the Everything but the Kitchen Sink integration."""
import voluptuous as vol
from homeassistant.components.device_tracker import DOMAIN as DEVICE_TRACKER_DOMAIN
from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE
from homeassistant.core import HomeAssistant, ServiceCall, ServiceResponse, callback
from homeassistant.helpers import config_validation as cv, service
from .const import DOMAIN
SCHEMA_SERVICE_TEST_SERVICE_1 = vol.Schema(
{
vol.Required("field_1"): vol.Coerce(int),
vol.Required("field_2"): vol.In(["off", "auto", "cool"]),
vol.Optional("field_3"): vol.Coerce(int),
vol.Optional("field_4"): vol.In(["forwards", "reverse"]),
}
)
SERVICE_TEST_SERVICE_1 = "test_service_1"
SERVICE_SET_TRACKER_LOCATION = "set_tracker_location"
SERVICE_SET_SCANNER_CONNECTED = "set_scanner_connected"
ATTR_ACCURACY = "accuracy"
ATTR_CONNECTED = "connected"
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Register services for the Kitchen Sink integration."""
@callback
def service_handler(call: ServiceCall | None = None) -> ServiceResponse:
"""Do nothing."""
return None
hass.services.async_register(
DOMAIN,
SERVICE_TEST_SERVICE_1,
service_handler,
SCHEMA_SERVICE_TEST_SERVICE_1,
description_placeholders={
"meep_1": "foo",
"meep_2": "bar",
"meep_3": "beer",
"meep_4": "milk",
"meep_5": "https://example.com",
},
)
service.async_register_platform_entity_service(
hass,
DOMAIN,
SERVICE_SET_TRACKER_LOCATION,
entity_domain=DEVICE_TRACKER_DOMAIN,
schema={
vol.Required(ATTR_LATITUDE): cv.latitude,
vol.Required(ATTR_LONGITUDE): cv.longitude,
vol.Required(ATTR_ACCURACY): vol.All(vol.Coerce(float), vol.Range(min=0)),
},
func="async_set_tracker_location",
)
service.async_register_platform_entity_service(
hass,
DOMAIN,
SERVICE_SET_SCANNER_CONNECTED,
entity_domain=DEVICE_TRACKER_DOMAIN,
schema={vol.Required(ATTR_CONNECTED): cv.boolean},
func="async_set_scanner_connected",
)
@@ -30,3 +30,44 @@ test_service_1:
options:
- "forward"
- "reverse"
set_tracker_location:
target:
entity:
integration: kitchen_sink
domain: device_tracker
fields:
latitude:
required: true
example: 52.379189
selector:
number:
min: -90
max: 90
step: any
longitude:
required: true
example: 4.899431
selector:
number:
min: -180
max: 180
step: any
accuracy:
required: true
example: 10
selector:
number:
min: 0
max: 10000
unit_of_measurement: m
set_scanner_connected:
target:
entity:
integration: kitchen_sink
domain: device_tracker
fields:
connected:
required: true
example: true
selector:
boolean:
@@ -135,6 +135,34 @@
}
},
"services": {
"set_scanner_connected": {
"description": "Sets the connected state of a demo scanner entity.",
"fields": {
"connected": {
"description": "Whether the device should be reported as connected.",
"name": "Connected"
}
},
"name": "Set scanner connected"
},
"set_tracker_location": {
"description": "Sets the location and accuracy of a demo tracker entity.",
"fields": {
"accuracy": {
"description": "Location accuracy in meters.",
"name": "Accuracy"
},
"latitude": {
"description": "Latitude of the new location.",
"name": "Latitude"
},
"longitude": {
"description": "Longitude of the new location.",
"name": "Longitude"
}
},
"name": "Set tracker location"
},
"test_service_1": {
"description": "Fake action for testing {meep_2}",
"fields": {
@@ -0,0 +1,57 @@
# serializer version: 1
# name: test_states
set({
StateSnapshot({
'attributes': ReadOnlyDict({
'editable': True,
'friendly_name': 'test home',
'icon': 'mdi:home',
'latitude': 32.87336,
'longitude': -117.22743,
'passive': False,
'persons': list([
]),
'radius': 100,
}),
'context': <ANY>,
'entity_id': 'zone.home',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
}),
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Demo scanner',
'in_zones': list([
'zone.home',
]),
'source_type': <SourceType.ROUTER: 'router'>,
}),
'context': <ANY>,
'entity_id': 'device_tracker.demo_scanner',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'home',
}),
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Demo tracker',
'gps_accuracy': 10,
'in_zones': list([
'zone.home',
]),
'latitude': 32.87336,
'longitude': -117.22743,
'source_type': <SourceType.GPS: 'gps'>,
}),
'context': <ANY>,
'entity_id': 'device_tracker.demo_tracker',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'home',
}),
})
# ---
@@ -0,0 +1,146 @@
"""The tests for the kitchen_sink device_tracker platform."""
from collections.abc import Generator
from unittest.mock import patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.device_tracker import ATTR_SOURCE_TYPE, SourceType
from homeassistant.components.kitchen_sink import DOMAIN
from homeassistant.components.kitchen_sink.services import (
ATTR_ACCURACY,
ATTR_CONNECTED,
SERVICE_SET_SCANNER_CONNECTED,
SERVICE_SET_TRACKER_LOCATION,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_GPS_ACCURACY,
ATTR_LATITUDE,
ATTR_LONGITUDE,
STATE_HOME,
STATE_NOT_HOME,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
TRACKER_ENTITY_ID = "device_tracker.demo_tracker"
SCANNER_ENTITY_ID = "device_tracker.demo_scanner"
@pytest.fixture
def device_tracker_only() -> Generator[None]:
"""Enable only the device_tracker platform."""
with patch(
"homeassistant.components.kitchen_sink.COMPONENTS_WITH_DEMO_PLATFORM",
[Platform.DEVICE_TRACKER],
):
yield
@pytest.fixture(autouse=True)
async def setup_comp(hass: HomeAssistant, device_tracker_only: None) -> None:
"""Set up demo component."""
hass.config.latitude = 32.87336
hass.config.longitude = -117.22743
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
await hass.async_block_till_done()
async def test_states(hass: HomeAssistant, snapshot: SnapshotAssertion) -> None:
"""Test the expected device_tracker entities are added."""
states = hass.states.async_all()
assert set(states) == snapshot
async def test_set_tracker_location(hass: HomeAssistant) -> None:
"""Test the set_tracker_location service updates tracker attributes."""
state = hass.states.get(TRACKER_ENTITY_ID)
assert state is not None
assert state.attributes[ATTR_SOURCE_TYPE] == SourceType.GPS
await hass.services.async_call(
DOMAIN,
SERVICE_SET_TRACKER_LOCATION,
{
ATTR_ENTITY_ID: TRACKER_ENTITY_ID,
ATTR_LATITUDE: 12.34,
ATTR_LONGITUDE: 56.78,
ATTR_ACCURACY: 42,
},
blocking=True,
)
state = hass.states.get(TRACKER_ENTITY_ID)
assert state.attributes[ATTR_LATITUDE] == 12.34
assert state.attributes[ATTR_LONGITUDE] == 56.78
assert state.attributes[ATTR_GPS_ACCURACY] == 42
assert state.state == STATE_NOT_HOME
async def test_set_scanner_connected(hass: HomeAssistant) -> None:
"""Test the set_scanner_connected service updates scanner state."""
state = hass.states.get(SCANNER_ENTITY_ID)
assert state is not None
assert state.state == STATE_HOME
assert state.attributes[ATTR_SOURCE_TYPE] == SourceType.ROUTER
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCANNER_CONNECTED,
{ATTR_ENTITY_ID: SCANNER_ENTITY_ID, ATTR_CONNECTED: False},
blocking=True,
)
state = hass.states.get(SCANNER_ENTITY_ID)
assert state.state == STATE_NOT_HOME
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCANNER_CONNECTED,
{ATTR_ENTITY_ID: SCANNER_ENTITY_ID, ATTR_CONNECTED: True},
blocking=True,
)
state = hass.states.get(SCANNER_ENTITY_ID)
assert state.state == STATE_HOME
async def test_set_tracker_location_on_scanner_raises(hass: HomeAssistant) -> None:
"""Calling set_tracker_location on the scanner surfaces an AttributeError.
The service is registered for the device_tracker domain and dispatches by
method name, so targeting the scanner (which has no async_set_tracker_location)
bubbles up the missing-attribute error from the entity.
"""
with pytest.raises(
AttributeError,
match="'DemoScanner' object has no attribute 'async_set_tracker_location'",
):
await hass.services.async_call(
DOMAIN,
SERVICE_SET_TRACKER_LOCATION,
{
ATTR_ENTITY_ID: SCANNER_ENTITY_ID,
ATTR_LATITUDE: 12.34,
ATTR_LONGITUDE: 56.78,
ATTR_ACCURACY: 42,
},
blocking=True,
)
async def test_set_scanner_connected_on_tracker_raises(hass: HomeAssistant) -> None:
"""Calling set_scanner_connected on the tracker surfaces an AttributeError."""
with pytest.raises(
AttributeError,
match="'DemoTracker' object has no attribute 'async_set_scanner_connected'",
):
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCANNER_CONNECTED,
{ATTR_ENTITY_ID: TRACKER_ENTITY_ID, ATTR_CONNECTED: False},
blocking=True,
)