mirror of
https://github.com/home-assistant/core.git
synced 2026-07-10 16:19:29 +01:00
Add UniFi Protect relay output switches via public API (#169201)
Co-authored-by: RaHehl <rahehl@users.noreply.github.com>
This commit is contained in:
Raphael Hehl
GitHub
@@ -52,9 +52,9 @@ DEVICES_THAT_ADOPT = {
|
||||
DEVICES_WITH_ENTITIES = DEVICES_THAT_ADOPT | {ModelType.NVR}
|
||||
DEVICES_FOR_SUBSCRIBE = DEVICES_WITH_ENTITIES | {ModelType.EVENT}
|
||||
|
||||
# Public API devices WebSocket: NVR (for arm_mode updates) and Siren
|
||||
# (for siren active-state updates).
|
||||
DEVICES_WS_SUBSCRIBED_MODELS = {ModelType.NVR, ModelType.SIREN}
|
||||
# Public API devices WebSocket: NVR (for arm_mode updates), Relay
|
||||
# (for relay output state updates), and Siren (for siren active-state updates).
|
||||
DEVICES_WS_SUBSCRIBED_MODELS = {ModelType.NVR, ModelType.RELAY, ModelType.SIREN}
|
||||
|
||||
MIN_REQUIRED_PROTECT_V = Version("6.0.0")
|
||||
OUTDATED_LOG_MESSAGE = (
|
||||
|
||||
@@ -19,6 +19,7 @@ from uiprotect.data import (
|
||||
ModelType,
|
||||
ProtectAdoptableDeviceModel,
|
||||
PTZPatrol,
|
||||
Relay,
|
||||
Siren,
|
||||
WSSubscriptionMessage,
|
||||
)
|
||||
@@ -84,6 +85,9 @@ class ProtectData:
|
||||
self._subscriptions: defaultdict[
|
||||
str, set[Callable[[ProtectDeviceType], None]]
|
||||
] = defaultdict(set)
|
||||
self._relay_subscriptions: defaultdict[str, set[Callable[[Relay], None]]] = (
|
||||
defaultdict(set)
|
||||
)
|
||||
self._siren_subscriptions: defaultdict[str, set[Callable[[Siren], None]]] = (
|
||||
defaultdict(set)
|
||||
)
|
||||
@@ -184,8 +188,10 @@ class ProtectData:
|
||||
|
||||
The API client pre-filters messages to the model types listed in
|
||||
DEVICES_WS_SUBSCRIBED_MODELS. NVR messages signal the private NVR so
|
||||
alarm entities pick up the new arm state. Siren messages dispatch
|
||||
the merged Siren object by mac so siren entities can refresh.
|
||||
alarm entities pick up the new arm state. Relay messages dispatch
|
||||
the merged Relay object by mac so relay-output entities can refresh.
|
||||
Siren messages dispatch the merged Siren object by mac so siren entities
|
||||
can refresh.
|
||||
"""
|
||||
new_obj = message.new_obj
|
||||
if new_obj is None:
|
||||
@@ -197,6 +203,14 @@ class ProtectData:
|
||||
if new_obj.model is ModelType.NVR:
|
||||
self._async_signal_device_update(self.api.bootstrap.nvr)
|
||||
return
|
||||
if new_obj.model is ModelType.RELAY:
|
||||
relay = cast(Relay, new_obj)
|
||||
mac = relay.mac
|
||||
if subscriptions := self._relay_subscriptions.get(mac):
|
||||
_LOGGER.debug("Updating relay: %s (%s)", relay.name, mac)
|
||||
for update_callback in subscriptions:
|
||||
update_callback(relay)
|
||||
return
|
||||
if new_obj.model is ModelType.SIREN:
|
||||
self._async_signal_siren_update(cast(Siren, new_obj))
|
||||
|
||||
@@ -372,6 +386,10 @@ class ProtectData:
|
||||
for device in self.get_by_types(DEVICES_THAT_ADOPT):
|
||||
self._async_signal_device_update(device)
|
||||
if self.api.has_public_bootstrap:
|
||||
for relay in self.api.public_bootstrap.relays.values():
|
||||
if subscriptions := self._relay_subscriptions.get(relay.mac):
|
||||
for subscription_callback in subscriptions:
|
||||
subscription_callback(relay)
|
||||
for siren in self.api.public_bootstrap.sirens.values():
|
||||
self._async_signal_siren_update(siren)
|
||||
|
||||
@@ -402,6 +420,23 @@ class ProtectData:
|
||||
if not self._subscriptions[mac]:
|
||||
del self._subscriptions[mac]
|
||||
|
||||
@callback
|
||||
def async_subscribe_relay(
|
||||
self, mac: str, update_callback: Callable[[Relay], None]
|
||||
) -> CALLBACK_TYPE:
|
||||
"""Add a callback subscriber for relay updates."""
|
||||
self._relay_subscriptions[mac].add(update_callback)
|
||||
return partial(self._async_unsubscribe_relay, mac, update_callback)
|
||||
|
||||
@callback
|
||||
def _async_unsubscribe_relay(
|
||||
self, mac: str, update_callback: Callable[[Relay], None]
|
||||
) -> None:
|
||||
"""Remove a relay callback subscriber."""
|
||||
self._relay_subscriptions[mac].remove(update_callback)
|
||||
if not self._relay_subscriptions[mac]:
|
||||
del self._relay_subscriptions[mac]
|
||||
|
||||
@callback
|
||||
def async_subscribe_siren(
|
||||
self, mac: str, update_callback: Callable[[Siren], None]
|
||||
|
||||
@@ -641,6 +641,9 @@
|
||||
"privacy_mode": {
|
||||
"name": "Privacy mode"
|
||||
},
|
||||
"relay_output": {
|
||||
"name": "Output {output_name}"
|
||||
},
|
||||
"ssh_enabled": {
|
||||
"name": "[%key:component::unifiprotect::entity::binary_sensor::ssh_enabled::name%]"
|
||||
},
|
||||
@@ -697,6 +700,9 @@
|
||||
"ptz_preset_not_found": {
|
||||
"message": "Could not find PTZ preset with name {preset_name} on camera {camera_name}"
|
||||
},
|
||||
"relay_not_available": {
|
||||
"message": "Relay is no longer available"
|
||||
},
|
||||
"service_error": {
|
||||
"message": "Error calling UniFi Protect service, check the logs for more details"
|
||||
},
|
||||
|
||||
@@ -5,22 +5,29 @@ from __future__ import annotations
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from functools import partial
|
||||
from typing import Any
|
||||
from typing import Any, Literal
|
||||
|
||||
from uiprotect.data import (
|
||||
Camera,
|
||||
ModelType,
|
||||
ProtectAdoptableDeviceModel,
|
||||
PublicRelayOutput,
|
||||
RecordingMode,
|
||||
Relay,
|
||||
RelayOutputState,
|
||||
VideoMode,
|
||||
)
|
||||
|
||||
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
|
||||
from homeassistant.const import EntityCategory
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
from homeassistant.helpers.restore_state import RestoreEntity
|
||||
|
||||
from .const import DEFAULT_ATTRIBUTION, DEFAULT_BRAND, DOMAIN
|
||||
from .data import ProtectData, ProtectDeviceType, UFPConfigEntry
|
||||
from .entity import (
|
||||
BaseProtectEntity,
|
||||
@@ -421,6 +428,12 @@ NVR_SWITCHES: tuple[ProtectSwitchEntityDescription, ...] = (
|
||||
),
|
||||
)
|
||||
|
||||
_RELAY_STATE_MAP: dict[RelayOutputState, bool] = {
|
||||
RelayOutputState.ON: True,
|
||||
RelayOutputState.OFF: False,
|
||||
RelayOutputState.OFF_OTP: False,
|
||||
}
|
||||
|
||||
_MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
|
||||
ModelType.CAMERA: CAMERA_SWITCHES,
|
||||
ModelType.LIGHT: LIGHT_SWITCHES,
|
||||
@@ -562,3 +575,119 @@ async def async_setup_entry(
|
||||
for switch in NVR_SWITCHES
|
||||
)
|
||||
async_add_entities(entities)
|
||||
|
||||
# Public API: relay output switches. Only available when the public
|
||||
# bootstrap has been primed (requires API key + supported NVR firmware).
|
||||
api = data.api
|
||||
if api.has_public_bootstrap:
|
||||
relay_entities: list[ProtectRelayOutputSwitch] = [
|
||||
ProtectRelayOutputSwitch(data, relay, output)
|
||||
for relay in api.public_bootstrap.relays.values()
|
||||
for output in relay.outputs
|
||||
]
|
||||
if relay_entities:
|
||||
async_add_entities(relay_entities)
|
||||
|
||||
|
||||
class ProtectRelayOutputSwitch(SwitchEntity):
|
||||
"""Switch entity for a single relay output channel (Public API).
|
||||
|
||||
The relay device and its outputs are exposed through UniFi Protect's
|
||||
public integration API and cached in :attr:`ProtectApiClient.public_bootstrap`.
|
||||
Each output channel is represented as its own switch entity; turning it
|
||||
on/off goes through :meth:`Relay.activate_output`.
|
||||
"""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
_attr_attribution = DEFAULT_ATTRIBUTION
|
||||
_attr_should_poll = False
|
||||
_attr_translation_key = "relay_output"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
data: ProtectData,
|
||||
relay: Relay,
|
||||
output: PublicRelayOutput,
|
||||
) -> None:
|
||||
"""Initialize the relay output switch."""
|
||||
self.data = data
|
||||
self._relay_id = relay.id
|
||||
self._relay_mac = relay.mac
|
||||
self._output_id = output.id
|
||||
self._attr_unique_id = f"{relay.mac}_relay_output_{output.id}"
|
||||
self._attr_translation_placeholders = {
|
||||
"output_name": output.name or str(output.id),
|
||||
}
|
||||
nvr = data.api.bootstrap.nvr
|
||||
self._attr_device_info = DeviceInfo(
|
||||
connections={(dr.CONNECTION_NETWORK_MAC, relay.mac)},
|
||||
identifiers={(DOMAIN, relay.mac)},
|
||||
manufacturer=DEFAULT_BRAND,
|
||||
name=relay.name,
|
||||
model="Relay",
|
||||
via_device=(DOMAIN, nvr.mac),
|
||||
)
|
||||
self._update_from_relay(relay)
|
||||
|
||||
@property
|
||||
def _relay(self) -> Relay | None:
|
||||
api = self.data.api
|
||||
if not api.has_public_bootstrap:
|
||||
return None
|
||||
return api.public_bootstrap.relays.get(self._relay_id)
|
||||
|
||||
@callback
|
||||
def _update_from_relay(self, relay: Relay) -> None:
|
||||
"""Refresh ``_attr_is_on`` and availability from the cached relay."""
|
||||
output = relay.get_output(self._output_id)
|
||||
if output is None:
|
||||
self._attr_available = False
|
||||
self._attr_is_on = None
|
||||
return
|
||||
self._attr_available = self.data.last_update_success
|
||||
self._attr_is_on = (
|
||||
_RELAY_STATE_MAP.get(output.state) if output.state is not None else None
|
||||
)
|
||||
|
||||
@callback
|
||||
def _async_updated(self, relay: Relay) -> None:
|
||||
"""Handle a public relay WS update for this relay."""
|
||||
prev_state = (self._attr_available, self._attr_is_on)
|
||||
self._update_from_relay(relay)
|
||||
# If the relay was removed from the bootstrap while the WS update
|
||||
# was in flight, mark unavailable so commands cannot succeed.
|
||||
if self._relay is None:
|
||||
self._attr_available = False
|
||||
if (self._attr_available, self._attr_is_on) != prev_state:
|
||||
self.async_write_ha_state()
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Subscribe to public relay WS updates dispatched by ProtectData."""
|
||||
await super().async_added_to_hass()
|
||||
self.async_on_remove(
|
||||
self.data.async_subscribe_relay(self._relay_mac, self._async_updated)
|
||||
)
|
||||
|
||||
async def _activate_output(self, state: Literal["on", "off"]) -> None:
|
||||
"""Send activate_output to the relay, raising if unavailable."""
|
||||
if (relay := self._relay) is None:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="relay_not_available",
|
||||
)
|
||||
if relay.get_output(self._output_id) is None:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="relay_not_available",
|
||||
)
|
||||
await relay.activate_output(self._output_id, state=state)
|
||||
|
||||
@async_ufp_instance_command
|
||||
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn the relay output on."""
|
||||
await self._activate_output("on")
|
||||
|
||||
@async_ufp_instance_command
|
||||
async def async_turn_off(self, **kwargs: Any) -> None:
|
||||
"""Turn the relay output off."""
|
||||
await self._activate_output("off")
|
||||
|
||||
@@ -43,6 +43,7 @@ def _make_public_bootstrap(arm_mode: Mock | None) -> Mock:
|
||||
pb = Mock(spec=PublicBootstrap)
|
||||
pb.arm_mode = arm_mode
|
||||
pb.arm_profiles = {}
|
||||
pb.relays = {}
|
||||
pb.sirens = {}
|
||||
return pb
|
||||
|
||||
|
||||
@@ -0,0 +1,465 @@
|
||||
"""Tests for the UniFi Protect relay (Public API) switch entities."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
import pytest
|
||||
from uiprotect.data import (
|
||||
ModelType,
|
||||
PublicBootstrap,
|
||||
PublicRelayOutput,
|
||||
Relay,
|
||||
RelayOutputState,
|
||||
)
|
||||
from uiprotect.exceptions import ClientError, NotAuthorized
|
||||
from uiprotect.websocket import WebsocketState
|
||||
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
SERVICE_TURN_OFF,
|
||||
SERVICE_TURN_ON,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
STATE_UNAVAILABLE,
|
||||
STATE_UNKNOWN,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from .utils import MockUFPFixture, init_entry
|
||||
|
||||
RELAY_ID = "relay-id-1"
|
||||
RELAY_MAC = "AA:BB:CC:DD:EE:01"
|
||||
RELAY_NAME = "Garage Relay"
|
||||
OUTPUT_ID = 1
|
||||
OUTPUT_NAME = "output1"
|
||||
|
||||
SWITCH_ENTITY_ID = "switch.garage_relay_output_output1"
|
||||
|
||||
|
||||
def _make_output(
|
||||
output_id: int = OUTPUT_ID,
|
||||
name: str | None = OUTPUT_NAME,
|
||||
state: RelayOutputState | None = RelayOutputState.OFF,
|
||||
) -> Mock:
|
||||
"""Build a mock :class:`PublicRelayOutput`."""
|
||||
output = Mock(spec=PublicRelayOutput)
|
||||
output.id = output_id
|
||||
output.name = name
|
||||
output.state = state
|
||||
return output
|
||||
|
||||
|
||||
def _make_relay(
|
||||
*,
|
||||
outputs: list[Mock] | None = None,
|
||||
) -> Mock:
|
||||
"""Build a mock :class:`Relay` whose ``activate_output`` is awaitable."""
|
||||
relay = Mock(spec=Relay)
|
||||
relay.id = RELAY_ID
|
||||
relay.mac = RELAY_MAC
|
||||
relay.name = RELAY_NAME
|
||||
relay.model = ModelType.RELAY
|
||||
relay.outputs = outputs if outputs is not None else [_make_output()]
|
||||
|
||||
def get_output(output_id: int) -> Mock | None:
|
||||
return next((o for o in relay.outputs if o.id == output_id), None)
|
||||
|
||||
relay.get_output = get_output
|
||||
relay.activate_output = AsyncMock()
|
||||
return relay
|
||||
|
||||
|
||||
def _make_public_bootstrap(relay: Mock | None) -> Mock:
|
||||
"""Build a public bootstrap mock holding the given relay."""
|
||||
pb = Mock(spec=PublicBootstrap)
|
||||
pb.relays = {relay.id: relay} if relay is not None else {}
|
||||
pb.arm_mode = None
|
||||
pb.arm_profiles = {}
|
||||
pb.sirens = {}
|
||||
return pb
|
||||
|
||||
|
||||
@pytest.fixture(name="ufp_with_relay")
|
||||
def _ufp_with_relay(ufp: MockUFPFixture) -> tuple[MockUFPFixture, Mock]:
|
||||
"""Configure ufp fixture with a single relay accessible via public API."""
|
||||
relay = _make_relay()
|
||||
ufp.api.has_public_bootstrap = True
|
||||
ufp.api.public_bootstrap = _make_public_bootstrap(relay)
|
||||
return ufp, relay
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Switch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_relay_switch_not_created_without_public_bootstrap(
|
||||
hass: HomeAssistant, ufp: MockUFPFixture
|
||||
) -> None:
|
||||
"""No relay output switch is created when public bootstrap is unavailable."""
|
||||
ufp.api.has_public_bootstrap = False
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
assert hass.states.get(SWITCH_ENTITY_ID) is None
|
||||
|
||||
|
||||
async def test_relay_switch_created_with_state(
|
||||
hass: HomeAssistant,
|
||||
entity_registry: er.EntityRegistry,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""Relay output switch is created and reflects the cached state."""
|
||||
ufp, relay = ufp_with_relay
|
||||
relay.outputs[0].state = RelayOutputState.ON
|
||||
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
entry = entity_registry.async_get(SWITCH_ENTITY_ID)
|
||||
assert entry is not None
|
||||
assert entry.unique_id == f"{RELAY_MAC}_relay_output_{OUTPUT_ID}"
|
||||
|
||||
state = hass.states.get(SWITCH_ENTITY_ID)
|
||||
assert state is not None
|
||||
assert state.state == STATE_ON
|
||||
|
||||
|
||||
async def test_relay_switch_off_otp_is_off(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""OFF_OTP (over-temperature protection) is treated as ``off``."""
|
||||
ufp, relay = ufp_with_relay
|
||||
relay.outputs[0].state = RelayOutputState.OFF_OTP
|
||||
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
state = hass.states.get(SWITCH_ENTITY_ID)
|
||||
assert state is not None
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
|
||||
async def test_relay_switch_unknown_state_is_unknown(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""Unknown relay state should leave the switch state as ``unknown``."""
|
||||
ufp, relay = ufp_with_relay
|
||||
relay.outputs[0].state = RelayOutputState.UNKNOWN
|
||||
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
state = hass.states.get(SWITCH_ENTITY_ID)
|
||||
assert state is not None
|
||||
# ``is_on`` is None while ``available`` is True → state is "unknown".
|
||||
# "unavailable" would mean the device is unreachable; UNKNOWN output state
|
||||
# means state data was received but cannot be interpreted.
|
||||
assert state.state == STATE_UNKNOWN
|
||||
|
||||
|
||||
async def test_relay_switch_turn_on_off(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""Calling ``turn_on``/``turn_off`` invokes the public-API helper."""
|
||||
ufp, relay = ufp_with_relay
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
await hass.services.async_call(
|
||||
Platform.SWITCH,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
|
||||
blocking=True,
|
||||
)
|
||||
relay.activate_output.assert_awaited_once_with(OUTPUT_ID, state="on")
|
||||
relay.activate_output.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
Platform.SWITCH,
|
||||
SERVICE_TURN_OFF,
|
||||
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
|
||||
blocking=True,
|
||||
)
|
||||
relay.activate_output.assert_awaited_once_with(OUTPUT_ID, state="off")
|
||||
|
||||
|
||||
async def test_relay_switch_state_updates_from_public_ws(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""A public devices WS update for the relay refreshes the switch state."""
|
||||
ufp, relay = ufp_with_relay
|
||||
relay.outputs[0].state = RelayOutputState.OFF
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
state = hass.states.get(SWITCH_ENTITY_ID)
|
||||
assert state is not None
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
relay.outputs[0].state = RelayOutputState.ON
|
||||
|
||||
mock_msg = Mock()
|
||||
mock_msg.changed_data = {}
|
||||
mock_msg.old_obj = relay
|
||||
mock_msg.new_obj = relay
|
||||
assert ufp.devices_ws_subscription is not None
|
||||
ufp.devices_ws_subscription(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(SWITCH_ENTITY_ID)
|
||||
assert state is not None
|
||||
assert state.state == STATE_ON
|
||||
|
||||
|
||||
async def test_relay_switch_creates_one_entity_per_output(
|
||||
hass: HomeAssistant,
|
||||
entity_registry: er.EntityRegistry,
|
||||
ufp: MockUFPFixture,
|
||||
) -> None:
|
||||
"""Multiple outputs on a single relay yield multiple switch entities."""
|
||||
relay = _make_relay(
|
||||
outputs=[
|
||||
_make_output(output_id=1, name="output1"),
|
||||
_make_output(output_id=2, name="output2"),
|
||||
],
|
||||
)
|
||||
ufp.api.has_public_bootstrap = True
|
||||
ufp.api.public_bootstrap = _make_public_bootstrap(relay)
|
||||
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
assert entity_registry.async_get("switch.garage_relay_output_output1") is not None
|
||||
assert entity_registry.async_get("switch.garage_relay_output_output2") is not None
|
||||
|
||||
|
||||
async def test_relay_switch_command_error_raises(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""``activate_output`` errors are surfaced as :class:`HomeAssistantError`."""
|
||||
ufp, relay = ufp_with_relay
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
relay.activate_output.side_effect = NotAuthorized("denied")
|
||||
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
Platform.SWITCH,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
async def test_relay_switch_client_error_raises(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""``ClientError`` from ``activate_output`` is wrapped as HomeAssistantError."""
|
||||
ufp, relay = ufp_with_relay
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
relay.activate_output.side_effect = ClientError("timeout")
|
||||
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
Platform.SWITCH,
|
||||
SERVICE_TURN_OFF,
|
||||
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
async def test_relay_switch_command_when_relay_gone(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""Command raises HomeAssistantError when the relay is no longer in bootstrap."""
|
||||
ufp, _relay = ufp_with_relay
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
# Remove relay from bootstrap after setup.
|
||||
ufp.api.public_bootstrap.relays = {}
|
||||
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
Platform.SWITCH,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
async def test_relay_switch_command_when_bootstrap_unavailable(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""Command raises HomeAssistantError when has_public_bootstrap is False."""
|
||||
ufp, _relay = ufp_with_relay
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
ufp.api.has_public_bootstrap = False
|
||||
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
Platform.SWITCH,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
async def test_relay_switch_ws_update_no_state_change(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""WS update with the same state does not trigger an unnecessary state write."""
|
||||
ufp, relay = ufp_with_relay
|
||||
relay.outputs[0].state = RelayOutputState.ON
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
assert hass.states.get(SWITCH_ENTITY_ID).state == STATE_ON # type: ignore[union-attr]
|
||||
|
||||
# Fire update with identical state — entity state must not change.
|
||||
mock_msg = Mock()
|
||||
mock_msg.changed_data = {}
|
||||
mock_msg.old_obj = relay
|
||||
mock_msg.new_obj = relay
|
||||
assert ufp.devices_ws_subscription is not None
|
||||
ufp.devices_ws_subscription(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get(SWITCH_ENTITY_ID).state == STATE_ON # type: ignore[union-attr]
|
||||
|
||||
|
||||
async def test_relay_switch_becomes_unavailable_when_relay_removed(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""Entity becomes unavailable when the relay disappears from the bootstrap."""
|
||||
ufp, relay = ufp_with_relay
|
||||
relay.outputs[0].state = RelayOutputState.OFF
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
# Drop the relay from the public bootstrap.
|
||||
ufp.api.public_bootstrap.relays = {}
|
||||
|
||||
# Send a WS update whose output list is still valid; the entity must still
|
||||
# become unavailable because _relay now resolves to None.
|
||||
relay2 = _make_relay()
|
||||
relay2.id = relay.id
|
||||
relay2.mac = relay.mac
|
||||
|
||||
mock_msg = Mock()
|
||||
mock_msg.changed_data = {}
|
||||
mock_msg.old_obj = relay2
|
||||
mock_msg.new_obj = relay2
|
||||
assert ufp.devices_ws_subscription is not None
|
||||
ufp.devices_ws_subscription(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(SWITCH_ENTITY_ID)
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
|
||||
async def test_relay_switch_availability_follows_websocket_state(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""Relay switch becomes unavailable on WS disconnect and recovers on reconnect."""
|
||||
ufp, relay = ufp_with_relay
|
||||
relay.outputs[0].state = RelayOutputState.ON
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
assert hass.states.get(SWITCH_ENTITY_ID).state == STATE_ON # type: ignore[union-attr]
|
||||
|
||||
assert ufp.ws_state_subscription is not None
|
||||
ufp.ws_state_subscription(WebsocketState.DISCONNECTED)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(SWITCH_ENTITY_ID)
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
ufp.ws_state_subscription(WebsocketState.CONNECTED)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(SWITCH_ENTITY_ID)
|
||||
assert state is not None
|
||||
assert state.state == STATE_ON
|
||||
|
||||
|
||||
async def test_relay_public_ws_message_with_none_new_obj(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""Public WS message with new_obj=None is silently ignored."""
|
||||
ufp, _ = ufp_with_relay
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
state_before = hass.states.get(SWITCH_ENTITY_ID)
|
||||
assert state_before is not None
|
||||
|
||||
mock_msg = Mock()
|
||||
mock_msg.new_obj = None
|
||||
|
||||
assert ufp.devices_ws_subscription is not None
|
||||
ufp.devices_ws_subscription(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Entity state must be unchanged.
|
||||
assert hass.states.get(SWITCH_ENTITY_ID) == state_before
|
||||
|
||||
|
||||
async def test_relay_switch_output_removed_from_relay_update(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""WS update where the output is no longer present marks the entity unavailable."""
|
||||
ufp, relay = ufp_with_relay
|
||||
relay.outputs[0].state = RelayOutputState.ON
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
assert hass.states.get(SWITCH_ENTITY_ID).state == STATE_ON # type: ignore[union-attr]
|
||||
|
||||
# Build a relay WS update that no longer contains any outputs.
|
||||
relay_no_outputs = _make_relay(outputs=[])
|
||||
relay_no_outputs.id = relay.id
|
||||
relay_no_outputs.mac = relay.mac
|
||||
|
||||
mock_msg = Mock()
|
||||
mock_msg.changed_data = {}
|
||||
mock_msg.old_obj = relay_no_outputs
|
||||
mock_msg.new_obj = relay_no_outputs
|
||||
|
||||
assert ufp.devices_ws_subscription is not None
|
||||
ufp.devices_ws_subscription(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(SWITCH_ENTITY_ID)
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
|
||||
async def test_relay_switch_command_when_output_gone(
|
||||
hass: HomeAssistant,
|
||||
ufp_with_relay: tuple[MockUFPFixture, Mock],
|
||||
) -> None:
|
||||
"""Command raises HomeAssistantError when the relay output channel is no longer present."""
|
||||
ufp, relay = ufp_with_relay
|
||||
await init_entry(hass, ufp, [])
|
||||
|
||||
# Remove all outputs from the relay so get_output returns None.
|
||||
relay.outputs = []
|
||||
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
Platform.SWITCH,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
|
||||
blocking=True,
|
||||
)
|
||||
Reference in New Issue
Block a user