starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-10-06 22:38:02 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add UniFi Protect relay input binary sensors (#179905)

Browse Source
This commit is contained in:
Igor Augustyński
2026-09-28 19:21:32 +02:00
committed by GitHub
parent 45eff7b838
commit d6829beb4b
7 changed files with 578 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifiprotect/binary_sensor.py
+122 -1
View File
@@ -7,10 +7,14 @@ from typing import cast, override
from uiprotect.data import (
NVR,
DeviceState,
Fob,
ModelType,
MountType,
ProtectAdoptableDeviceModel,
PublicRelayInput,
Relay,
RelayInputState,
Sensor,
SmartDetectObjectType,
)
@@ -28,9 +32,15 @@ from homeassistant.components.binary_sensor import (
)
from homeassistant.const import EntityCategory, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.entity_platform import (
AddConfigEntryEntitiesCallback,
async_get_current_platform,
)
from .const import DEFAULT_ATTRIBUTION, DEFAULT_BRAND, DOMAIN
from .data import ProtectData, ProtectDeviceType, UFPConfigEntry
from .entity import (
BaseProtectEntity,
@@ -49,6 +59,11 @@ from .entity import (
_KEY_DOOR = "door"
PARALLEL_UPDATES = 0
_RELAY_INPUT_STATE_MAP: dict[RelayInputState, bool] = {
RelayInputState.ON: True,
RelayInputState.OFF: False,
}
def _async_motion_sensor_enabled_public(obj: PublicDeviceModel) -> bool:
# Mirrors Sensor.is_motion_sensor_enabled over the public API.
@@ -695,6 +710,93 @@ class ProtectEventBinarySensor(EventEntityMixin, BinarySensorEntity):
self._async_event_with_immediate_end()
class ProtectRelayInputBinarySensor(BinarySensorEntity):
"""Binary sensor for a single relay input channel (Public API)."""
_attr_has_entity_name = True
_attr_attribution = DEFAULT_ATTRIBUTION
_attr_should_poll = False
_attr_translation_key = "relay_input"
def __init__(
self,
data: ProtectData,
relay: Relay,
relay_input: PublicRelayInput,
) -> None:
"""Initialize the relay input binary sensor."""
self.data = data
self._relay_id = relay.id
self._relay_mac = relay.mac
self._input_id = relay_input.id
self._attr_unique_id = f"{relay.mac}_relay_input_{relay_input.id}"
self._attr_translation_placeholders = {
"input_name": relay_input.name or str(relay_input.id),
}
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_id=data.nvr_device_id,
)
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:
relay_input = next(
(
relay_input
for relay_input in relay.inputs
if relay_input.id == self._input_id
),
None,
)
if (
relay_input is None
or relay.state is not DeviceState.CONNECTED
or not self.data.last_public_update_success
):
self._attr_available = False
self._attr_is_on = None
return
self._attr_available = True
self._attr_is_on = (
_RELAY_INPUT_STATE_MAP.get(relay_input.state)
if relay_input.state is not None
else None
)
@callback
def _async_updated(self, _obj: PublicDeviceModel | None) -> None:
"""Refresh state from the public bootstrap cache."""
prev_state = (self._attr_available, self._attr_is_on)
if (relay := self._relay) is None:
self._attr_available = False
self._attr_is_on = None
else:
self._update_from_relay(relay)
if (self._attr_available, self._attr_is_on) != prev_state:
self.async_write_ha_state()
@override
async def async_added_to_hass(self) -> None:
"""Subscribe to public relay updates."""
await super().async_added_to_hass()
self.async_on_remove(
self.data.async_subscribe_public(self._relay_mac, self._async_updated)
)
self._async_updated(None)
MODEL_DESCRIPTIONS_WITH_CLASS = (
(_MODEL_DESCRIPTIONS, ProtectDeviceBinarySensor),
(_MOUNTABLE_MODEL_DESCRIPTIONS, MountableProtectDeviceBinarySensor),
@@ -785,6 +887,7 @@ async def async_setup_entry(
) -> None:
"""Set up binary sensors for UniFi Protect integration."""
data = entry.runtime_data
platform = async_get_current_platform()
@callback
def _add_new_public_device(device: PublicDeviceModel) -> None:
@@ -808,6 +911,24 @@ async def async_setup_entry(
for description in FOB_BINARY_SENSORS
)
@callback
def _add_relay_inputs(relay: Relay) -> None:
live_unique_ids = {entity.unique_id for entity in platform.entities.values()}
async_add_entities(
[
ProtectRelayInputBinarySensor(data, relay, relay_input)
for relay_input in relay.inputs
if f"{relay.mac}_relay_input_{relay_input.id}" not in live_unique_ids
]
)
entry.async_on_unload(
async_dispatcher_connect(hass, data.relay_signal, _add_relay_inputs)
)
if api.has_public_bootstrap:
for relay in api.public_bootstrap.relays.values():
_add_relay_inputs(relay)
# Everything below is driven by the private bootstrap, which public-only
# entries do not have.
if api.is_public_only:
homeassistant/components/unifiprotect/const.py
+1
View File
@@ -103,6 +103,7 @@ CONNECTION_MODE_API_KEY_ONLY = "api_key_only"
DISPATCH_ADD = "add_device"
DISPATCH_ADOPT = "adopt_device"
DISPATCH_CHANNELS = "new_camera_channels"
DISPATCH_RELAYS = "new_relay_channels"
DISPATCH_PUBLIC_ADD = "public_add_device"
EVENT_TYPE_FINGERPRINT_IDENTIFIED: Final = "identified"
homeassistant/components/unifiprotect/data.py
+12 -1
View File
@@ -21,6 +21,7 @@ from uiprotect.data import (
ProtectAdoptableDeviceModel,
PTZPatrol,
PublicDeviceModel,
Relay,
WSAction,
WSSubscriptionMessage,
)
@@ -48,6 +49,7 @@ from .const import (
DISPATCH_ADOPT,
DISPATCH_CHANNELS,
DISPATCH_PUBLIC_ADD,
DISPATCH_RELAYS,
DOMAIN,
)
from .utils import async_get_devices_by_type
@@ -155,6 +157,7 @@ class ProtectData:
self.adopt_signal = _async_dispatch_id(entry, DISPATCH_ADOPT)
self.add_signal = _async_dispatch_id(entry, DISPATCH_ADD)
self.channels_signal = _async_dispatch_id(entry, DISPATCH_CHANNELS)
self.relay_signal = _async_dispatch_id(entry, DISPATCH_RELAYS)
self.public_add_signal = _async_dispatch_id(entry, DISPATCH_PUBLIC_ADD)
# PTZ patrol cache: camera_id -> list of patrols
self.ptz_patrols: dict[str, list[PTZPatrol]] = {}
@@ -423,6 +426,12 @@ class ProtectData:
if message.action is WSAction.ADD:
self._async_dispatch_new_public_device(new_obj)
self._async_signal_public_update(new_obj.mac, new_obj)
if (
isinstance(new_obj, Relay)
and self.api.has_public_bootstrap
and new_obj.id in self.api.public_bootstrap.relays
):
async_dispatcher_send(self._hass, self.relay_signal, new_obj)
@callback
def _async_reenumerate_camera_on_public_change(
@@ -528,13 +537,15 @@ class ProtectData:
return
self._async_process_public_updates()
# A device that appeared during the gap gets no add frame, so re-offer
# everything; the dispatch helper drops what platforms already know.
# everything. Platforms and dispatch helpers drop what they already know.
if self.api.has_public_bootstrap:
for public in list(self.api.public_bootstrap.cameras.values()):
async_dispatcher_send(self._hass, self.channels_signal, public)
for device in list(self.api.public_bootstrap.all_devices()):
if isinstance(device, PublicDeviceModel):
self._async_dispatch_new_public_device(device)
for relay in list(self.api.public_bootstrap.relays.values()):
async_dispatcher_send(self._hass, self.relay_signal, relay)
@callback
def _async_signal_nvr_update(self) -> None:
homeassistant/components/unifiprotect/icons.json
+3
View File
@@ -126,6 +126,9 @@
"person_detected": {
"default": "mdi:walk"
},
"relay_input": {
"default": "mdi:import"
},
"siren_detected": {
"default": "mdi:alarm-bell"
},
homeassistant/components/unifiprotect/strings.json
+3
View File
@@ -295,6 +295,9 @@
"person_detected": {
"name": "Person detected"
},
"relay_input": {
"name": "Input {input_name}"
},
"siren_detected": {
"name": "Siren detected"
},
homeassistant/components/unifiprotect/switch.py
+20 -18
View File
@@ -553,10 +553,24 @@ async def async_setup_entry(
entities += _make_entities(ProtectPrivacyModeSwitch, _PRIVACY_DESCRIPTIONS)
async_add_entities(entities)
relay_output_unique_ids: set[str] = set()
@callback
def _add_relay_outputs(relay: Relay) -> None:
entities: list[ProtectRelayOutputSwitch] = []
for output in relay.outputs:
unique_id = f"{relay.mac}_relay_output_{output.id}"
if unique_id in relay_output_unique_ids:
continue
relay_output_unique_ids.add(unique_id)
entities.append(ProtectRelayOutputSwitch(data, relay, output))
if entities:
async_add_entities(entities)
@callback
def _add_new_public_device(device: PublicDeviceModel) -> None:
if isinstance(device, Relay):
async_add_entities(_relay_output_switches(data, device))
_add_relay_outputs(device)
return
async_add_entities(
async_all_device_entities(
@@ -569,6 +583,9 @@ async def async_setup_entry(
entry.async_on_unload(
async_dispatcher_connect(hass, data.public_add_signal, _add_new_public_device)
)
entry.async_on_unload(
async_dispatcher_connect(hass, data.relay_signal, _add_relay_outputs)
)
entities: list[BaseProtectEntity] = []
entities += _make_entities(ProtectSwitch, _MODEL_DESCRIPTIONS)
entities += _make_entities(ProtectPrivacyModeSwitch, _PRIVACY_DESCRIPTIONS)
@@ -584,24 +601,9 @@ async def async_setup_entry(
)
async_add_entities(entities)
# Relays exist only in the public API; a relay adopted later arrives
# through the public add signal in either mode.
if api.has_public_bootstrap:
relay_entities = [
entity
for relay in api.public_bootstrap.relays.values()
for entity in _relay_output_switches(data, relay)
]
if relay_entities:
async_add_entities(relay_entities)
@callback
def _relay_output_switches(
data: ProtectData, relay: Relay
) -> list[ProtectRelayOutputSwitch]:
"""Build one switch per output channel of a relay."""
return [ProtectRelayOutputSwitch(data, relay, output) for output in relay.outputs]
for relay in api.public_bootstrap.relays.values():
_add_relay_outputs(relay)
class ProtectRelayOutputSwitch(SwitchEntity):
tests/components/unifiprotect/test_relay.py
+417 -1
View File
@@ -1,4 +1,4 @@
"""Tests for the UniFi Protect relay (Public API) switch entities."""
"""Tests for UniFi Protect relay entities from the Public API."""
from collections.abc import Callable, Coroutine
from typing import Any
@@ -8,8 +8,11 @@ import pytest
from uiprotect.data import (
DeviceState,
ModelType,
PublicBootstrap,
PublicRelayInput,
PublicRelayOutput,
Relay,
RelayInputState,
RelayOutputState,
WSAction,
)
@@ -43,8 +46,11 @@ RELAY_MAC = "AA:BB:CC:DD:EE:01"
RELAY_NAME = "Garage Relay"
OUTPUT_ID = 1
OUTPUT_NAME = "output1"
INPUT_ID = 1
INPUT_NAME = "input1"
SWITCH_ENTITY_ID = "switch.garage_relay_output_output1"
BINARY_SENSOR_ENTITY_ID = "binary_sensor.garage_relay_input_input1"
def _make_output(
@@ -60,9 +66,23 @@ def _make_output(
return output
def _make_input(
input_id: int = INPUT_ID,
name: str | None = INPUT_NAME,
state: RelayInputState | None = RelayInputState.OFF,
) -> Mock:
"""Build a mock :class:`PublicRelayInput`."""
relay_input = Mock(spec=PublicRelayInput)
relay_input.id = input_id
relay_input.name = name
relay_input.state = state
return relay_input
def _make_relay(
*,
outputs: list[Mock] | None = None,
inputs: list[Mock] | None = None,
state: DeviceState = DeviceState.CONNECTED,
) -> Mock:
"""Build a mock :class:`Relay` whose ``activate_output`` is awaitable."""
@@ -73,6 +93,7 @@ def _make_relay(
relay.model = ModelType.RELAY
relay.state = state
relay.outputs = outputs if outputs is not None else [_make_output()]
relay.inputs = inputs if inputs is not None else [_make_input()]
def get_output(output_id: int) -> Mock | None:
return next((o for o in relay.outputs if o.id == output_id), None)
@@ -87,6 +108,24 @@ def _make_public_bootstrap(relay: Mock | None) -> Mock:
return make_public_bootstrap(relays={relay.id: relay} if relay is not None else {})
def _make_real_relay(ufp: MockUFPFixture) -> Relay:
"""Build a relay using the pinned uiprotect public model."""
return Relay.from_unifi_dict(
api=ufp.api,
id=RELAY_ID,
modelKey="relay",
state="CONNECTED",
mac=RELAY_MAC,
name=RELAY_NAME,
ledSettings={"isEnabled": True},
outputs=[],
inputs=[
{"id": 0, "name": "Garage Door Fully Open", "state": "off"},
{"id": 1, "name": "Garage Door Fully Closed", "state": "off"},
],
)
@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."""
@@ -96,6 +135,383 @@ def _ufp_with_relay(ufp: MockUFPFixture) -> tuple[MockUFPFixture, Mock]:
return ufp, relay
def _send_relay_update(ufp: MockUFPFixture, relay: Mock) -> None:
"""Dispatch a public devices websocket update for a relay."""
message = Mock()
message.changed_data = {}
message.old_obj = relay
message.new_obj = relay
assert ufp.devices_ws_subscription is not None
ufp.devices_ws_subscription(message)
async def test_relay_input_not_created_without_public_bootstrap(
hass: HomeAssistant, ufp: MockUFPFixture
) -> None:
"""Relay inputs require the public bootstrap."""
ufp.api.has_public_bootstrap = False
await init_entry(hass, ufp, [])
assert hass.states.get(BINARY_SENSOR_ENTITY_ID) is None
@pytest.mark.parametrize(
("inputs", "entity_ids"),
[
pytest.param(
[],
[],
id="no_inputs",
),
pytest.param(
[_make_input(input_id=4, name="Door")],
["binary_sensor.garage_relay_input_door"],
id="one_input",
),
pytest.param(
[
_make_input(input_id=4, name="Door"),
_make_input(input_id=7, name=None),
],
[
"binary_sensor.garage_relay_input_door",
"binary_sensor.garage_relay_input_7",
],
id="multiple_inputs",
),
],
)
async def test_relay_input_enumeration_names_and_unique_ids(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
ufp: MockUFPFixture,
inputs: list[Mock],
entity_ids: list[str],
) -> None:
"""Create one stably identified entity per named or unnamed input."""
relay = _make_relay(inputs=inputs)
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = _make_public_bootstrap(relay)
await init_entry(hass, ufp, [])
entries = [entity_registry.async_get(entity_id) for entity_id in entity_ids]
assert all(entry is not None for entry in entries)
assert [entry.unique_id for entry in entries if entry is not None] == [
f"{RELAY_MAC}_relay_input_{relay_input.id}" for relay_input in inputs
]
assert len(
[
entry
for entry in entity_registry.entities.values()
if entry.unique_id.startswith(f"{RELAY_MAC}_relay_input_")
]
) == len(inputs)
async def test_relay_input_unique_id_does_not_depend_on_name(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
ufp_with_relay: tuple[MockUFPFixture, Mock],
) -> None:
"""Configured input names affect only entity names, not unique IDs."""
ufp, relay = ufp_with_relay
relay.inputs[0].name = "Side door"
await init_entry(hass, ufp, [])
entry = entity_registry.async_get("binary_sensor.garage_relay_input_side_door")
assert entry is not None
assert entry.unique_id == f"{RELAY_MAC}_relay_input_{INPUT_ID}"
assert entry.original_name == "Input Side door"
@pytest.mark.parametrize(
("input_state", "expected_state"),
[
pytest.param(RelayInputState.ON, STATE_ON, id="on"),
pytest.param(RelayInputState.OFF, STATE_OFF, id="off"),
pytest.param(RelayInputState.UNKNOWN, STATE_UNKNOWN, id="unknown"),
pytest.param(None, STATE_UNKNOWN, id="none"),
],
)
async def test_relay_input_initial_state(
hass: HomeAssistant,
ufp_with_relay: tuple[MockUFPFixture, Mock],
input_state: RelayInputState | None,
expected_state: str,
) -> None:
"""Map sustained public input states to binary sensor states."""
ufp, relay = ufp_with_relay
relay.inputs[0].state = input_state
await init_entry(hass, ufp, [])
state = hass.states.get(BINARY_SENSOR_ENTITY_ID)
assert state is not None
assert state.state == expected_state
assert state.attributes.get("device_class") is None
async def test_relay_input_transitions_both_ways_from_public_ws(
hass: HomeAssistant,
ufp_with_relay: tuple[MockUFPFixture, Mock],
) -> None:
"""Public devices websocket updates drive both input transitions."""
ufp, relay = ufp_with_relay
await init_entry(hass, ufp, [])
state = hass.states.get(BINARY_SENSOR_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
relay.inputs[0].state = RelayInputState.ON
_send_relay_update(ufp, relay)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
relay.inputs[0].state = RelayInputState.OFF
_send_relay_update(ufp, relay)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
async def test_relay_input_update_preserves_other_input(
hass: HomeAssistant, ufp: MockUFPFixture
) -> None:
"""A realistic full inputs update retains both input channels."""
relay = _make_real_relay(ufp)
ufp.api.has_public_bootstrap = True
public_bootstrap = PublicBootstrap(relays={relay.id: relay})
ufp.api.public_bootstrap = public_bootstrap
await init_entry(hass, ufp, [])
_model, new_obj, old_obj = public_bootstrap.process_devices_ws_message(
ufp.api,
{
"type": "update",
"item": {
"id": RELAY_ID,
"modelKey": "relay",
"inputs": [
{"id": 0, "name": "Garage Door Fully Open", "state": "on"},
{"id": 1, "name": "Garage Door Fully Closed", "state": "off"},
],
},
},
)
assert isinstance(new_obj, Relay)
message = Mock()
message.changed_data = {"inputs": new_obj.inputs}
message.old_obj = old_obj
message.new_obj = new_obj
assert ufp.devices_ws_subscription is not None
ufp.devices_ws_subscription(message)
await hass.async_block_till_done()
assert [relay_input.id for relay_input in new_obj.inputs] == [0, 1]
first_state = hass.states.get(
"binary_sensor.garage_relay_input_garage_door_fully_open"
)
second_state = hass.states.get(
"binary_sensor.garage_relay_input_garage_door_fully_closed"
)
assert first_state is not None
assert second_state is not None
assert first_state.state == STATE_ON
assert second_state.state == STATE_OFF
async def test_relay_entities_created_for_relay_added_from_public_ws(
hass: HomeAssistant, ufp: MockUFPFixture
) -> None:
"""A relay added after setup creates its input and output entities."""
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = _make_public_bootstrap(None)
await init_entry(hass, ufp, [])
relay = _make_relay()
ufp.api.public_bootstrap.relays[relay.id] = relay
_send_relay_update(ufp, relay)
await hass.async_block_till_done()
assert hass.states.get(BINARY_SENSOR_ENTITY_ID) is not None
assert hass.states.get(SWITCH_ENTITY_ID) is not None
async def test_relay_input_created_when_added_after_setup(
hass: HomeAssistant, ufp: MockUFPFixture
) -> None:
"""An input added to an existing relay creates a binary sensor."""
relay = _make_relay(inputs=[])
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = _make_public_bootstrap(relay)
await init_entry(hass, ufp, [])
assert hass.states.get(BINARY_SENSOR_ENTITY_ID) is None
relay.inputs = [_make_input()]
_send_relay_update(ufp, relay)
await hass.async_block_till_done()
assert hass.states.get(BINARY_SENSOR_ENTITY_ID) is not None
async def test_relay_input_unavailable_when_relay_disconnected(
hass: HomeAssistant,
ufp_with_relay: tuple[MockUFPFixture, Mock],
) -> None:
"""A disconnected relay makes its input unavailable."""
ufp, relay = ufp_with_relay
await init_entry(hass, ufp, [])
relay.state = DeviceState.DISCONNECTED
_send_relay_update(ufp, relay)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_ENTITY_ID)
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def test_relay_input_devices_ws_disconnect_reconnect_resync(
hass: HomeAssistant,
ufp_with_relay: tuple[MockUFPFixture, Mock],
) -> None:
"""Input availability and state recover from the resynced public bootstrap."""
ufp, relay = ufp_with_relay
await init_entry(hass, ufp, [])
assert ufp.devices_ws_state_subscription is not None
ufp.devices_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_ENTITY_ID)
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def resync_public_bootstrap() -> Mock:
relay.inputs[0].state = RelayInputState.ON
return ufp.api.public_bootstrap
ufp.api.update_public.side_effect = resync_public_bootstrap
ufp.devices_ws_state_subscription(WebsocketState.CONNECTED)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
ufp.api.update_public.assert_awaited()
async def test_public_only_relay_channels_resignaled_after_reconnect(
hass: HomeAssistant,
ufp_public_only: MockUFPFixture,
setup_public_only: Callable[[], Coroutine[Any, Any, None]],
) -> None:
"""A public-only reconnect re-offers channels on an existing relay."""
relay = _make_relay(inputs=[])
public_bootstrap = ufp_public_only.api.public_bootstrap
public_bootstrap.relays = {relay.id: relay}
await setup_public_only()
assert hass.states.get(BINARY_SENSOR_ENTITY_ID) is None
async def resync_public_bootstrap() -> Mock:
relay.inputs = [_make_input()]
return public_bootstrap
ufp_public_only.api.update_public.side_effect = resync_public_bootstrap
assert ufp_public_only.devices_ws_state_subscription is not None
ufp_public_only.devices_ws_state_subscription(WebsocketState.DISCONNECTED)
ufp_public_only.devices_ws_state_subscription(WebsocketState.CONNECTED)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
@pytest.mark.parametrize(
"remove_channel",
[
pytest.param(
lambda ufp, _relay: setattr(ufp.api.public_bootstrap, "relays", {}),
id="relay_removed",
),
pytest.param(
lambda _ufp, relay: setattr(relay, "inputs", []),
id="input_removed",
),
],
)
async def test_relay_input_unavailable_when_channel_missing(
hass: HomeAssistant,
ufp_with_relay: tuple[MockUFPFixture, Mock],
remove_channel: Callable[[MockUFPFixture, Mock], None],
) -> None:
"""A removed relay or input makes the existing entity unavailable."""
ufp, relay = ufp_with_relay
relay.inputs[0].state = RelayInputState.ON
await init_entry(hass, ufp, [])
remove_channel(ufp, relay)
_send_relay_update(ufp, relay)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_ENTITY_ID)
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def test_relay_input_uses_same_relay_and_nvr_device(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
ufp_with_relay: tuple[MockUFPFixture, Mock],
) -> None:
"""Input and output share the relay device linked to the NVR."""
ufp, _relay = ufp_with_relay
await init_entry(hass, ufp, [])
input_entry = entity_registry.async_get(BINARY_SENSOR_ENTITY_ID)
output_entry = entity_registry.async_get(SWITCH_ENTITY_ID)
assert input_entry is not None
assert output_entry is not None
assert input_entry.device_id == output_entry.device_id
relay_device = device_registry.async_get(input_entry.device_id)
nvr_device = device_registry.async_get_device_by_identifier(
(DOMAIN, ufp.api.bootstrap.nvr.mac), ufp.entry.entry_id
)
assert relay_device is not None
assert nvr_device is not None
assert relay_device.connections == {(dr.CONNECTION_NETWORK_MAC, RELAY_MAC.lower())}
assert relay_device.identifiers == {(DOMAIN, RELAY_MAC)}
assert relay_device.manufacturer == "Ubiquiti"
assert relay_device.model == "Relay"
assert relay_device.via_device_id == nvr_device.id
async def test_relay_input_ignores_events_ws_health(
hass: HomeAssistant,
ufp_with_relay: tuple[MockUFPFixture, Mock],
) -> None:
"""Events websocket health does not affect sustained relay input state."""
ufp, relay = ufp_with_relay
relay.inputs[0].state = RelayInputState.ON
await init_entry(hass, ufp, [])
assert ufp.events_ws_state_subscription is not None
ufp.events_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
# ---------------------------------------------------------------------------
# Switch
# ---------------------------------------------------------------------------