feat: Einstellungen, Neustart, Diagnose-Sensoren und Diagnose-Download
This commit is contained in:
@@ -13,7 +13,14 @@ from .api import NinlightClient
|
|||||||
from .const import DEFAULT_TOKEN, DOMAIN
|
from .const import DEFAULT_TOKEN, DOMAIN
|
||||||
from .coordinator import NinlightConfigEntry, NinlightCoordinator
|
from .coordinator import NinlightConfigEntry, NinlightCoordinator
|
||||||
|
|
||||||
PLATFORMS: list[Platform] = [Platform.LIGHT]
|
PLATFORMS: list[Platform] = [
|
||||||
|
Platform.BUTTON,
|
||||||
|
Platform.LIGHT,
|
||||||
|
Platform.NUMBER,
|
||||||
|
Platform.SELECT,
|
||||||
|
Platform.SENSOR,
|
||||||
|
Platform.SWITCH,
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(hass: HomeAssistant, entry: NinlightConfigEntry) -> bool:
|
async def async_setup_entry(hass: HomeAssistant, entry: NinlightConfigEntry) -> bool:
|
||||||
|
|||||||
@@ -0,0 +1,34 @@
|
|||||||
|
"""Reboot button of a ninlight lamp. A factory reset is deliberately not offered."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from homeassistant.components.button import ButtonDeviceClass, ButtonEntity
|
||||||
|
from homeassistant.const import EntityCategory
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||||
|
|
||||||
|
from .coordinator import NinlightConfigEntry, NinlightCoordinator
|
||||||
|
from .entity import NinlightEntity
|
||||||
|
|
||||||
|
PARALLEL_UPDATES = 0
|
||||||
|
|
||||||
|
|
||||||
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
entry: NinlightConfigEntry,
|
||||||
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||||
|
) -> None:
|
||||||
|
async_add_entities([NinlightRebootButton(entry.runtime_data)])
|
||||||
|
|
||||||
|
|
||||||
|
class NinlightRebootButton(NinlightEntity, ButtonEntity):
|
||||||
|
"""Restart the lamp."""
|
||||||
|
|
||||||
|
_attr_device_class = ButtonDeviceClass.RESTART
|
||||||
|
_attr_entity_category = EntityCategory.CONFIG
|
||||||
|
|
||||||
|
def __init__(self, coordinator: NinlightCoordinator) -> None:
|
||||||
|
super().__init__(coordinator, "reboot")
|
||||||
|
|
||||||
|
async def async_press(self) -> None:
|
||||||
|
await self.coordinator.async_call(lambda client: client.reboot())
|
||||||
@@ -0,0 +1,27 @@
|
|||||||
|
"""Diagnostics download for ninlight lamps."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from homeassistant.components.diagnostics import async_redact_data
|
||||||
|
from homeassistant.const import CONF_HOST, CONF_TOKEN
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
|
||||||
|
from .coordinator import NinlightConfigEntry
|
||||||
|
|
||||||
|
TO_REDACT = {CONF_TOKEN, CONF_HOST, "ssid", "mac", "ip", "sta_ip"}
|
||||||
|
|
||||||
|
|
||||||
|
async def async_get_config_entry_diagnostics(
|
||||||
|
hass: HomeAssistant, entry: NinlightConfigEntry
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
coordinator = entry.runtime_data
|
||||||
|
return async_redact_data(
|
||||||
|
{
|
||||||
|
"entry": {"data": dict(entry.data), "options": dict(entry.options)},
|
||||||
|
"state": coordinator.data.state,
|
||||||
|
"info": coordinator.data.info,
|
||||||
|
},
|
||||||
|
TO_REDACT,
|
||||||
|
)
|
||||||
@@ -0,0 +1,69 @@
|
|||||||
|
"""Fade and transition settings of a ninlight lamp."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
from homeassistant.components.number import (
|
||||||
|
NumberDeviceClass,
|
||||||
|
NumberEntity,
|
||||||
|
NumberEntityDescription,
|
||||||
|
NumberMode,
|
||||||
|
)
|
||||||
|
from homeassistant.const import EntityCategory, UnitOfTime
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||||
|
|
||||||
|
from .coordinator import NinlightConfigEntry, NinlightCoordinator
|
||||||
|
from .entity import NinlightEntity
|
||||||
|
|
||||||
|
PARALLEL_UPDATES = 0
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, kw_only=True)
|
||||||
|
class NinlightNumberDescription(NumberEntityDescription):
|
||||||
|
"""A millisecond setting stored under `key` in the lamp state."""
|
||||||
|
|
||||||
|
entity_category: EntityCategory = EntityCategory.CONFIG
|
||||||
|
device_class: NumberDeviceClass = NumberDeviceClass.DURATION
|
||||||
|
native_unit_of_measurement: str = UnitOfTime.MILLISECONDS
|
||||||
|
native_min_value: float = 0
|
||||||
|
native_step: float = 50
|
||||||
|
mode: NumberMode = NumberMode.BOX
|
||||||
|
|
||||||
|
|
||||||
|
NUMBERS = (
|
||||||
|
NinlightNumberDescription(key="fade_in", translation_key="fade_in", native_max_value=10_000),
|
||||||
|
NinlightNumberDescription(key="fade_out", translation_key="fade_out", native_max_value=10_000),
|
||||||
|
NinlightNumberDescription(
|
||||||
|
key="transition_default", translation_key="transition_default", native_max_value=60_000
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
entry: NinlightConfigEntry,
|
||||||
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||||
|
) -> None:
|
||||||
|
async_add_entities(NinlightNumber(entry.runtime_data, description) for description in NUMBERS)
|
||||||
|
|
||||||
|
|
||||||
|
class NinlightNumber(NinlightEntity, NumberEntity):
|
||||||
|
"""One millisecond setting."""
|
||||||
|
|
||||||
|
entity_description: NinlightNumberDescription
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self, coordinator: NinlightCoordinator, description: NinlightNumberDescription
|
||||||
|
) -> None:
|
||||||
|
super().__init__(coordinator, description.key)
|
||||||
|
self.entity_description = description
|
||||||
|
|
||||||
|
@property
|
||||||
|
def native_value(self) -> int | None:
|
||||||
|
value = self.coordinator.data.state.get(self.entity_description.key)
|
||||||
|
return int(value) if value is not None else None
|
||||||
|
|
||||||
|
async def async_set_native_value(self, value: float) -> None:
|
||||||
|
await self.coordinator.async_set_state(**{self.entity_description.key: int(value)})
|
||||||
@@ -0,0 +1,43 @@
|
|||||||
|
"""Power-on behaviour of a ninlight lamp."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from homeassistant.components.select import SelectEntity
|
||||||
|
from homeassistant.const import EntityCategory
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||||
|
|
||||||
|
from .coordinator import NinlightConfigEntry, NinlightCoordinator
|
||||||
|
from .entity import NinlightEntity
|
||||||
|
|
||||||
|
PARALLEL_UPDATES = 0
|
||||||
|
|
||||||
|
# POB_NAMES in the firmware's nl_api.c
|
||||||
|
POWER_ON_OPTIONS = ["last", "on", "off"]
|
||||||
|
|
||||||
|
|
||||||
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
entry: NinlightConfigEntry,
|
||||||
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||||
|
) -> None:
|
||||||
|
async_add_entities([NinlightPowerOnSelect(entry.runtime_data)])
|
||||||
|
|
||||||
|
|
||||||
|
class NinlightPowerOnSelect(NinlightEntity, SelectEntity):
|
||||||
|
"""What the lamp does when power returns."""
|
||||||
|
|
||||||
|
_attr_entity_category = EntityCategory.CONFIG
|
||||||
|
_attr_translation_key = "power_on_behaviour"
|
||||||
|
_attr_options = POWER_ON_OPTIONS
|
||||||
|
|
||||||
|
def __init__(self, coordinator: NinlightCoordinator) -> None:
|
||||||
|
super().__init__(coordinator, "power_on_behaviour")
|
||||||
|
|
||||||
|
@property
|
||||||
|
def current_option(self) -> str | None:
|
||||||
|
value = self.coordinator.data.state.get("power_on_behaviour")
|
||||||
|
return value if value in POWER_ON_OPTIONS else None
|
||||||
|
|
||||||
|
async def async_select_option(self, option: str) -> None:
|
||||||
|
await self.coordinator.async_set_state(power_on_behaviour=option)
|
||||||
@@ -0,0 +1,135 @@
|
|||||||
|
"""Diagnostic sensors of a ninlight lamp."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Callable
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from homeassistant.components.sensor import (
|
||||||
|
SensorDeviceClass,
|
||||||
|
SensorEntity,
|
||||||
|
SensorEntityDescription,
|
||||||
|
SensorStateClass,
|
||||||
|
)
|
||||||
|
from homeassistant.const import (
|
||||||
|
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
|
||||||
|
EntityCategory,
|
||||||
|
UnitOfInformation,
|
||||||
|
)
|
||||||
|
from homeassistant.core import HomeAssistant, callback
|
||||||
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
|
from .coordinator import NinlightConfigEntry, NinlightCoordinator, NinlightData
|
||||||
|
from .entity import NinlightEntity
|
||||||
|
|
||||||
|
PARALLEL_UPDATES = 0
|
||||||
|
|
||||||
|
# A new boot time is only published when it moves by more than this: uptime and the local
|
||||||
|
# clock drift a few seconds against each other on every poll.
|
||||||
|
BOOT_TIME_TOLERANCE = timedelta(seconds=60)
|
||||||
|
|
||||||
|
|
||||||
|
def _wifi(data: NinlightData) -> dict[str, Any]:
|
||||||
|
wifi = data.info.get("wifi")
|
||||||
|
return wifi if isinstance(wifi, dict) else {}
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, kw_only=True)
|
||||||
|
class NinlightSensorDescription(SensorEntityDescription):
|
||||||
|
"""A sensor reading one value out of the lamp data."""
|
||||||
|
|
||||||
|
value_fn: Callable[[NinlightData], Any]
|
||||||
|
entity_category: EntityCategory = EntityCategory.DIAGNOSTIC
|
||||||
|
|
||||||
|
|
||||||
|
SENSORS = (
|
||||||
|
NinlightSensorDescription(
|
||||||
|
key="rssi",
|
||||||
|
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
|
||||||
|
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
|
||||||
|
state_class=SensorStateClass.MEASUREMENT,
|
||||||
|
value_fn=lambda data: _wifi(data).get("rssi"),
|
||||||
|
),
|
||||||
|
NinlightSensorDescription(
|
||||||
|
key="ip_address",
|
||||||
|
translation_key="ip_address",
|
||||||
|
entity_registry_enabled_default=False,
|
||||||
|
value_fn=lambda data: _wifi(data).get("ip"),
|
||||||
|
),
|
||||||
|
NinlightSensorDescription(
|
||||||
|
key="free_heap",
|
||||||
|
translation_key="free_heap",
|
||||||
|
device_class=SensorDeviceClass.DATA_SIZE,
|
||||||
|
native_unit_of_measurement=UnitOfInformation.BYTES,
|
||||||
|
state_class=SensorStateClass.MEASUREMENT,
|
||||||
|
entity_registry_enabled_default=False,
|
||||||
|
value_fn=lambda data: data.info.get("free_heap"),
|
||||||
|
),
|
||||||
|
NinlightSensorDescription(
|
||||||
|
key="last_ota_error",
|
||||||
|
translation_key="last_ota_error",
|
||||||
|
entity_registry_enabled_default=False,
|
||||||
|
value_fn=lambda data: data.info.get("last_ota_error"),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
entry: NinlightConfigEntry,
|
||||||
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||||
|
) -> None:
|
||||||
|
coordinator = entry.runtime_data
|
||||||
|
entities: list[SensorEntity] = [NinlightSensor(coordinator, d) for d in SENSORS]
|
||||||
|
entities.append(NinlightBootTimeSensor(coordinator))
|
||||||
|
async_add_entities(entities)
|
||||||
|
|
||||||
|
|
||||||
|
class NinlightSensor(NinlightEntity, SensorEntity):
|
||||||
|
"""A plain diagnostic value."""
|
||||||
|
|
||||||
|
entity_description: NinlightSensorDescription
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self, coordinator: NinlightCoordinator, description: NinlightSensorDescription
|
||||||
|
) -> None:
|
||||||
|
super().__init__(coordinator, description.key)
|
||||||
|
self.entity_description = description
|
||||||
|
|
||||||
|
@property
|
||||||
|
def native_value(self) -> Any:
|
||||||
|
return self.entity_description.value_fn(self.coordinator.data)
|
||||||
|
|
||||||
|
|
||||||
|
class NinlightBootTimeSensor(NinlightEntity, SensorEntity):
|
||||||
|
"""When the lamp last started - a timestamp instead of an ever-growing uptime."""
|
||||||
|
|
||||||
|
_attr_device_class = SensorDeviceClass.TIMESTAMP
|
||||||
|
_attr_entity_category = EntityCategory.DIAGNOSTIC
|
||||||
|
_attr_entity_registry_enabled_default = False
|
||||||
|
_attr_translation_key = "boot_time"
|
||||||
|
|
||||||
|
def __init__(self, coordinator: NinlightCoordinator) -> None:
|
||||||
|
super().__init__(coordinator, "boot_time")
|
||||||
|
self._boot: datetime | None = None
|
||||||
|
self._update_boot()
|
||||||
|
|
||||||
|
def _update_boot(self) -> None:
|
||||||
|
uptime = self.coordinator.data.info.get("uptime")
|
||||||
|
if not isinstance(uptime, int | float):
|
||||||
|
return
|
||||||
|
boot = dt_util.utcnow() - timedelta(seconds=uptime)
|
||||||
|
if self._boot is None or abs(boot - self._boot) > BOOT_TIME_TOLERANCE:
|
||||||
|
self._boot = boot
|
||||||
|
|
||||||
|
@callback
|
||||||
|
def _handle_coordinator_update(self) -> None:
|
||||||
|
self._update_boot()
|
||||||
|
super()._handle_coordinator_update()
|
||||||
|
|
||||||
|
@property
|
||||||
|
def native_value(self) -> datetime | None:
|
||||||
|
return self._boot
|
||||||
@@ -0,0 +1,43 @@
|
|||||||
|
"""Wake-on-change switch of a ninlight lamp."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from homeassistant.components.switch import SwitchEntity
|
||||||
|
from homeassistant.const import EntityCategory
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||||
|
|
||||||
|
from .coordinator import NinlightConfigEntry, NinlightCoordinator
|
||||||
|
from .entity import NinlightEntity
|
||||||
|
|
||||||
|
PARALLEL_UPDATES = 0
|
||||||
|
|
||||||
|
|
||||||
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
entry: NinlightConfigEntry,
|
||||||
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||||
|
) -> None:
|
||||||
|
async_add_entities([NinlightWakeSwitch(entry.runtime_data)])
|
||||||
|
|
||||||
|
|
||||||
|
class NinlightWakeSwitch(NinlightEntity, SwitchEntity):
|
||||||
|
"""Turn the lamp on when its colour, temperature or brightness is changed."""
|
||||||
|
|
||||||
|
_attr_entity_category = EntityCategory.CONFIG
|
||||||
|
_attr_translation_key = "wake_on_change"
|
||||||
|
|
||||||
|
def __init__(self, coordinator: NinlightCoordinator) -> None:
|
||||||
|
super().__init__(coordinator, "wake_on_change")
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_on(self) -> bool:
|
||||||
|
return bool(self.coordinator.data.state.get("wake_on_change"))
|
||||||
|
|
||||||
|
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||||
|
await self.coordinator.async_set_state(wake_on_change=True)
|
||||||
|
|
||||||
|
async def async_turn_off(self, **kwargs: Any) -> None:
|
||||||
|
await self.coordinator.async_set_state(wake_on_change=False)
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
"""Reboot button."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN
|
||||||
|
from homeassistant.components.button import SERVICE_PRESS
|
||||||
|
from homeassistant.const import ATTR_ENTITY_ID
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
|
||||||
|
from tests.helpers import entity_id_for
|
||||||
|
|
||||||
|
|
||||||
|
async def test_press_reboots(hass: HomeAssistant, init_integration, mock_client: MagicMock) -> None:
|
||||||
|
entity_id = entity_id_for(hass, "button", "reboot")
|
||||||
|
assert hass.states.get(entity_id).attributes["device_class"] == "restart"
|
||||||
|
await hass.services.async_call(
|
||||||
|
BUTTON_DOMAIN, SERVICE_PRESS, {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||||
|
)
|
||||||
|
mock_client.reboot.assert_awaited_once_with()
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
"""Diagnostics download must not leak secrets or network identity."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from pytest_homeassistant_custom_component.components.diagnostics import (
|
||||||
|
get_diagnostics_for_config_entry,
|
||||||
|
)
|
||||||
|
from pytest_homeassistant_custom_component.typing import ClientSessionGenerator
|
||||||
|
|
||||||
|
from tests_unit.samples import HOST_RGB, MAC_RGB, TOKEN
|
||||||
|
|
||||||
|
|
||||||
|
async def test_diagnostics_redacts_secrets(
|
||||||
|
hass: HomeAssistant, hass_client: ClientSessionGenerator, init_integration
|
||||||
|
) -> None:
|
||||||
|
diagnostics = await get_diagnostics_for_config_entry(hass, hass_client, init_integration)
|
||||||
|
text = json.dumps(diagnostics)
|
||||||
|
for secret in (TOKEN, MAC_RGB, "ninIOT", HOST_RGB):
|
||||||
|
assert secret not in text, secret
|
||||||
|
assert diagnostics["info"]["device_type"] == "gu10_rgb"
|
||||||
|
assert diagnostics["state"]["brightness"] == 37
|
||||||
@@ -0,0 +1,49 @@
|
|||||||
|
"""Fade and transition settings."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
from homeassistant.components.number import ATTR_VALUE, SERVICE_SET_VALUE
|
||||||
|
from homeassistant.components.number import DOMAIN as NUMBER_DOMAIN
|
||||||
|
from homeassistant.const import ATTR_ENTITY_ID, ATTR_UNIT_OF_MEASUREMENT, UnitOfTime
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.helpers import entity_registry as er
|
||||||
|
from homeassistant.helpers.entity import EntityCategory
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from tests.helpers import entity_id_for
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("key", "initial", "new", "maximum"),
|
||||||
|
[
|
||||||
|
("fade_in", "450", 1300, 10000),
|
||||||
|
("fade_out", "1250", 3350, 10000),
|
||||||
|
("transition_default", "700", 42050, 60000),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
async def test_number_reads_and_writes_its_field(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
init_integration,
|
||||||
|
mock_client: MagicMock,
|
||||||
|
key: str,
|
||||||
|
initial: str,
|
||||||
|
new: int,
|
||||||
|
maximum: int,
|
||||||
|
) -> None:
|
||||||
|
entity_id = entity_id_for(hass, "number", key)
|
||||||
|
state = hass.states.get(entity_id)
|
||||||
|
assert state.state == initial
|
||||||
|
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == UnitOfTime.MILLISECONDS
|
||||||
|
assert state.attributes["max"] == maximum
|
||||||
|
assert er.async_get(hass).async_get(entity_id).entity_category is EntityCategory.CONFIG
|
||||||
|
|
||||||
|
await hass.services.async_call(
|
||||||
|
NUMBER_DOMAIN,
|
||||||
|
SERVICE_SET_VALUE,
|
||||||
|
{ATTR_ENTITY_ID: entity_id, ATTR_VALUE: new},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_client.set_state.assert_awaited_once_with(**{key: new})
|
||||||
|
assert hass.states.get(entity_id).state == str(new)
|
||||||
@@ -0,0 +1,39 @@
|
|||||||
|
"""Power-on behaviour."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
from homeassistant.components.select import ATTR_OPTION, SERVICE_SELECT_OPTION
|
||||||
|
from homeassistant.components.select import DOMAIN as SELECT_DOMAIN
|
||||||
|
from homeassistant.const import ATTR_ENTITY_ID
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
|
||||||
|
from tests.helpers import entity_id_for
|
||||||
|
|
||||||
|
|
||||||
|
async def test_select_reads_and_writes(
|
||||||
|
hass: HomeAssistant, init_integration, mock_client: MagicMock
|
||||||
|
) -> None:
|
||||||
|
entity_id = entity_id_for(hass, "select", "power_on_behaviour")
|
||||||
|
state = hass.states.get(entity_id)
|
||||||
|
assert state.state == "last"
|
||||||
|
assert state.attributes["options"] == ["last", "on", "off"]
|
||||||
|
await hass.services.async_call(
|
||||||
|
SELECT_DOMAIN,
|
||||||
|
SERVICE_SELECT_OPTION,
|
||||||
|
{ATTR_ENTITY_ID: entity_id, ATTR_OPTION: "off"},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_client.set_state.assert_awaited_once_with(power_on_behaviour="off")
|
||||||
|
assert hass.states.get(entity_id).state == "off"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_unknown_value_from_lamp_is_unknown(
|
||||||
|
hass: HomeAssistant, mock_config_entry, mock_client: MagicMock, lamp_state
|
||||||
|
) -> None:
|
||||||
|
lamp_state["power_on_behaviour"] = "fade"
|
||||||
|
mock_config_entry.add_to_hass(hass)
|
||||||
|
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
assert hass.states.get(entity_id_for(hass, "select", "power_on_behaviour")).state == "unknown"
|
||||||
@@ -0,0 +1,83 @@
|
|||||||
|
"""Diagnostic sensors."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import timedelta
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from freezegun.api import FrozenDateTimeFactory
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.helpers import entity_registry as er
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
from pytest_homeassistant_custom_component.common import MockConfigEntry, async_fire_time_changed
|
||||||
|
|
||||||
|
from tests.helpers import entity_id_for
|
||||||
|
|
||||||
|
DISABLED_KEYS = ("ip_address", "free_heap", "last_ota_error", "boot_time")
|
||||||
|
|
||||||
|
|
||||||
|
async def _enable_all(
|
||||||
|
hass: HomeAssistant, entry: MockConfigEntry, freezer: FrozenDateTimeFactory
|
||||||
|
) -> None:
|
||||||
|
registry = er.async_get(hass)
|
||||||
|
for key in DISABLED_KEYS:
|
||||||
|
registry.async_update_entity(entity_id_for(hass, "sensor", key), disabled_by=None)
|
||||||
|
await hass.config_entries.async_reload(entry.entry_id)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
# HA 2026.9: registry.async_update_entity(disabled_by=None) also arms the config entry's
|
||||||
|
# own debounced reload (config_entries.RELOAD_AFTER_UPDATE_DELAY = 30 s), independent of
|
||||||
|
# the explicit reload above. Left alone, that second reload fires mid-test - typically
|
||||||
|
# right on top of the first _poll() below - and silently replaces the boot-time entity,
|
||||||
|
# defeating the jitter-tolerance assertion. Drain it here so the entities are settled
|
||||||
|
# before the test takes its "first" reading.
|
||||||
|
freezer.tick(timedelta(seconds=31))
|
||||||
|
async_fire_time_changed(hass)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
|
||||||
|
async def _poll(hass: HomeAssistant, freezer: FrozenDateTimeFactory, seconds: int) -> None:
|
||||||
|
freezer.tick(timedelta(seconds=seconds))
|
||||||
|
async_fire_time_changed(hass)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_signal_enabled_others_disabled_by_default(
|
||||||
|
hass: HomeAssistant, init_integration
|
||||||
|
) -> None:
|
||||||
|
registry = er.async_get(hass)
|
||||||
|
assert hass.states.get(entity_id_for(hass, "sensor", "rssi")).state == "-61"
|
||||||
|
for key in DISABLED_KEYS:
|
||||||
|
entry = registry.async_get(entity_id_for(hass, "sensor", key))
|
||||||
|
assert entry.disabled_by is er.RegistryEntryDisabler.INTEGRATION, key
|
||||||
|
|
||||||
|
|
||||||
|
async def test_diagnostic_values(
|
||||||
|
hass: HomeAssistant, init_integration: MockConfigEntry, freezer
|
||||||
|
) -> None:
|
||||||
|
await _enable_all(hass, init_integration, freezer)
|
||||||
|
assert hass.states.get(entity_id_for(hass, "sensor", "ip_address")).state == "192.168.2.22"
|
||||||
|
assert hass.states.get(entity_id_for(hass, "sensor", "free_heap")).state == "119872"
|
||||||
|
assert hass.states.get(entity_id_for(hass, "sensor", "last_ota_error")).state == "unknown"
|
||||||
|
expected_boot = dt_util.utcnow() - timedelta(seconds=129832)
|
||||||
|
boot = dt_util.parse_datetime(hass.states.get(entity_id_for(hass, "sensor", "boot_time")).state)
|
||||||
|
assert abs((boot - expected_boot).total_seconds()) < 2
|
||||||
|
|
||||||
|
|
||||||
|
async def test_boot_time_ignores_jitter_but_follows_reboots(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
init_integration: MockConfigEntry,
|
||||||
|
lamp_info: dict[str, Any],
|
||||||
|
freezer: FrozenDateTimeFactory,
|
||||||
|
) -> None:
|
||||||
|
await _enable_all(hass, init_integration, freezer)
|
||||||
|
entity_id = entity_id_for(hass, "sensor", "boot_time")
|
||||||
|
first = hass.states.get(entity_id).state
|
||||||
|
|
||||||
|
lamp_info["uptime"] += 31 + 20 # clock drift of 20 s: below the 60 s threshold
|
||||||
|
await _poll(hass, freezer, 31)
|
||||||
|
assert hass.states.get(entity_id).state == first
|
||||||
|
|
||||||
|
lamp_info["uptime"] = 12 # the lamp rebooted
|
||||||
|
await _poll(hass, freezer, 31)
|
||||||
|
rebooted = dt_util.parse_datetime(hass.states.get(entity_id).state)
|
||||||
|
assert abs((rebooted - (dt_util.utcnow() - timedelta(seconds=12))).total_seconds()) < 2
|
||||||
@@ -0,0 +1,33 @@
|
|||||||
|
"""Wake-on-change switch."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
|
||||||
|
from homeassistant.const import (
|
||||||
|
ATTR_ENTITY_ID,
|
||||||
|
SERVICE_TURN_OFF,
|
||||||
|
SERVICE_TURN_ON,
|
||||||
|
STATE_OFF,
|
||||||
|
STATE_ON,
|
||||||
|
)
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
|
||||||
|
from tests.helpers import entity_id_for
|
||||||
|
|
||||||
|
|
||||||
|
async def test_switch_reads_and_writes(
|
||||||
|
hass: HomeAssistant, init_integration, mock_client: MagicMock
|
||||||
|
) -> None:
|
||||||
|
entity_id = entity_id_for(hass, "switch", "wake_on_change")
|
||||||
|
assert hass.states.get(entity_id).state == STATE_ON
|
||||||
|
await hass.services.async_call(
|
||||||
|
SWITCH_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||||
|
)
|
||||||
|
mock_client.set_state.assert_awaited_once_with(wake_on_change=False)
|
||||||
|
assert hass.states.get(entity_id).state == STATE_OFF
|
||||||
|
await hass.services.async_call(
|
||||||
|
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||||
|
)
|
||||||
|
mock_client.set_state.assert_awaited_with(wake_on_change=True)
|
||||||
Reference in New Issue
Block a user