mirror of
https://github.com/home-assistant/core.git
synced 2026-05-27 03:05:50 +02:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 6f2b83411d |
@@ -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,
|
||||
)
|
||||
Reference in New Issue
Block a user