feat: reject writes when a device's remote control is disabled
Devices with a /remotectrl href already surface it as a read-only "Smart Control" binary sensor, but writes weren't checking it before now. async_send_command now blocks every write (any platform) with a ServiceValidationError telling the user to enable remote control via the appliance's manual, ahead of any per-description validate_fn.
This commit is contained in:
Marc Billow
@@ -37,6 +37,26 @@ _LOGGER = logging.getLogger(__name__)
|
||||
|
||||
_SEED_PATH = ['device', '0']
|
||||
|
||||
_REMOTE_CONTROL_DISABLED_MESSAGE = (
|
||||
"Remote control is turned off on this device. Check your appliance's "
|
||||
"manual for how to enable remote control before Home Assistant can "
|
||||
"control it."
|
||||
)
|
||||
|
||||
|
||||
def _remote_control_disabled(resources: dict) -> bool:
|
||||
"""True only when the device explicitly reports remote control off.
|
||||
Mirrors REMOTE_CONTROL_GENERIC/_VS_FALLBACK's href/field pair in
|
||||
registry/capabilities/common.py. Fails open (False) when neither href
|
||||
is present -- most device types don't report this capability at all."""
|
||||
generic = resources.get('/remotectrl/0')
|
||||
if generic is not None:
|
||||
return not bool(generic.get('value'))
|
||||
fallback = resources.get('/remotectrl/vs/0')
|
||||
if fallback is not None:
|
||||
return str(fallback.get('x.com.samsung.da.remoteControlEnabled')).lower() != 'true'
|
||||
return False
|
||||
|
||||
|
||||
class _NoOpDescriptor:
|
||||
"""StateCache requires a descriptor with an on_observation hook. This
|
||||
@@ -515,16 +535,21 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
||||
A description-level validate_fn (currently SwitchDesc only) runs
|
||||
here rather than per-platform, so rejecting a write with a
|
||||
user-facing message -- as opposed to write_fn's silent no-op below
|
||||
-- is available to every platform for free."""
|
||||
-- is available to every platform for free. The remote-control
|
||||
check runs first and applies to every platform unconditionally,
|
||||
ahead of any description-specific validate_fn."""
|
||||
desc = bound_entity.desc
|
||||
write_fn = getattr(desc, 'write_fn', None)
|
||||
if write_fn is None:
|
||||
return
|
||||
href = bound_entity.href
|
||||
rep = self._cache.get(href or '') or {}
|
||||
resources = self._cache.snapshot()
|
||||
if _remote_control_disabled(resources):
|
||||
raise ServiceValidationError(_REMOTE_CONTROL_DISABLED_MESSAGE)
|
||||
validate_fn = getattr(desc, 'validate_fn', None)
|
||||
if validate_fn is not None:
|
||||
error = validate_fn(payload, rep, self._cache.snapshot())
|
||||
error = validate_fn(payload, rep, resources)
|
||||
if error:
|
||||
raise ServiceValidationError(error)
|
||||
result = write_fn(payload, rep, href)
|
||||
|
||||
@@ -7,12 +7,15 @@ from unittest.mock import AsyncMock, patch
|
||||
import cbor2
|
||||
import pytest
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ServiceValidationError
|
||||
from homeassistant.helpers import issue_registry as ir
|
||||
|
||||
from custom_components.localthings.const import (
|
||||
CONF_HOST, DOMAIN, SUMMARY_INTERVAL_S,
|
||||
)
|
||||
from custom_components.localthings.coordinator import LocalThingsCoordinator
|
||||
from custom_components.localthings.coordinator import (
|
||||
LocalThingsCoordinator, _remote_control_disabled,
|
||||
)
|
||||
from custom_components.localthings.observe import MODE_OBSERVE, MODE_POLL, PUSH_HEALTH_WINDOW_S
|
||||
|
||||
from .conftest import ENTRY_DATA, MOCK_SERIAL
|
||||
@@ -634,3 +637,136 @@ async def test_write_marks_href_pending_before_post(
|
||||
await coordinator.async_send_command(bound, 5)
|
||||
|
||||
assert coordinator._observe._settle_until.get('/test/vs/0') is not None
|
||||
|
||||
|
||||
class TestRemoteControlDisabled:
|
||||
"""_remote_control_disabled mirrors REMOTE_CONTROL_GENERIC/_VS_FALLBACK's
|
||||
OCF-standard-preferred read (registry/capabilities/common.py) directly
|
||||
on a raw resources snapshot."""
|
||||
|
||||
def test_vs_fallback_href_true_is_enabled(self):
|
||||
resources = {'/remotectrl/vs/0': {'x.com.samsung.da.remoteControlEnabled': 'true'}}
|
||||
assert _remote_control_disabled(resources) is False
|
||||
|
||||
def test_vs_fallback_href_false_is_disabled(self):
|
||||
resources = {'/remotectrl/vs/0': {'x.com.samsung.da.remoteControlEnabled': 'false'}}
|
||||
assert _remote_control_disabled(resources) is True
|
||||
|
||||
def test_generic_href_true_is_enabled(self):
|
||||
resources = {'/remotectrl/0': {'value': True}}
|
||||
assert _remote_control_disabled(resources) is False
|
||||
|
||||
def test_generic_href_false_is_disabled(self):
|
||||
resources = {'/remotectrl/0': {'value': False}}
|
||||
assert _remote_control_disabled(resources) is True
|
||||
|
||||
def test_generic_href_wins_when_both_present(self):
|
||||
resources = {
|
||||
'/remotectrl/0': {'value': True},
|
||||
'/remotectrl/vs/0': {'x.com.samsung.da.remoteControlEnabled': 'false'},
|
||||
}
|
||||
assert _remote_control_disabled(resources) is False
|
||||
|
||||
def test_fails_open_when_capability_absent(self):
|
||||
assert _remote_control_disabled({}) is False
|
||||
|
||||
|
||||
async def test_send_command_blocked_when_remote_control_disabled(
|
||||
hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
|
||||
) -> None:
|
||||
"""A device reporting remote control off must reject every write with
|
||||
a user-facing message, ahead of write_fn's silent-no-op path and any
|
||||
description-level validate_fn."""
|
||||
from custom_components.localthings.registry.discovery import BoundEntity
|
||||
from custom_components.localthings.registry.entities import NumberDesc
|
||||
|
||||
fake = mock_coordinator_observe_session
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
|
||||
coordinator._cache.apply_rep(
|
||||
'/remotectrl/vs/0',
|
||||
{'x.com.samsung.da.remoteControlEnabled': 'false'},
|
||||
source='test',
|
||||
)
|
||||
|
||||
def _write_fn(payload, rep, href):
|
||||
return (['some', 'path'], {'value': payload})
|
||||
|
||||
desc = NumberDesc(key='test', field='value', write_fn=_write_fn)
|
||||
bound = BoundEntity(href='/test/vs/0', capability=coordinator.bound[0].capability, desc=desc)
|
||||
|
||||
posted = False
|
||||
|
||||
def _post(*a, **k):
|
||||
nonlocal posted
|
||||
posted = True
|
||||
return (0x44, b'')
|
||||
|
||||
with patch.object(fake, 'subscribe'):
|
||||
fake.post = _post
|
||||
with pytest.raises(ServiceValidationError, match="Remote control is turned off"):
|
||||
await coordinator.async_send_command(bound, 5)
|
||||
|
||||
assert posted is False
|
||||
|
||||
|
||||
async def test_send_command_allowed_when_remote_control_enabled(
|
||||
hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
|
||||
) -> None:
|
||||
"""The same write goes through once remote control reports enabled."""
|
||||
from custom_components.localthings.registry.discovery import BoundEntity
|
||||
from custom_components.localthings.registry.entities import NumberDesc
|
||||
|
||||
fake = mock_coordinator_observe_session
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
|
||||
coordinator._cache.apply_rep(
|
||||
'/remotectrl/vs/0',
|
||||
{'x.com.samsung.da.remoteControlEnabled': 'true'},
|
||||
source='test',
|
||||
)
|
||||
|
||||
def _write_fn(payload, rep, href):
|
||||
return (['some', 'path'], {'value': payload})
|
||||
|
||||
desc = NumberDesc(key='test', field='value', write_fn=_write_fn)
|
||||
bound = BoundEntity(href='/test/vs/0', capability=coordinator.bound[0].capability, desc=desc)
|
||||
|
||||
with patch.object(fake, 'subscribe'):
|
||||
fake.post = lambda *a, **k: (0x44, b'')
|
||||
await coordinator.async_send_command(bound, 5)
|
||||
|
||||
assert coordinator._observe._settle_until.get('/test/vs/0') is not None
|
||||
|
||||
|
||||
async def test_send_command_remote_control_check_precedes_validate_fn(
|
||||
hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
|
||||
) -> None:
|
||||
"""When both a disabled remote-control state and a failing validate_fn
|
||||
apply to the same write, the remote-control message wins -- it's the
|
||||
more actionable of the two and is checked first."""
|
||||
from custom_components.localthings.registry.discovery import BoundEntity
|
||||
from custom_components.localthings.registry.entities import SwitchDesc
|
||||
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
|
||||
coordinator._cache.apply_rep(
|
||||
'/remotectrl/vs/0',
|
||||
{'x.com.samsung.da.remoteControlEnabled': 'false'},
|
||||
source='test',
|
||||
)
|
||||
|
||||
def _write_fn(payload, rep, href=None):
|
||||
return (['some', 'path'], {'value': payload})
|
||||
|
||||
def _validate_fn(payload, rep, resources):
|
||||
return "always rejected"
|
||||
|
||||
desc = SwitchDesc(key='test', field='value', write_fn=_write_fn, validate_fn=_validate_fn)
|
||||
bound = BoundEntity(href='/test/vs/0', capability=coordinator.bound[0].capability, desc=desc)
|
||||
|
||||
with pytest.raises(ServiceValidationError, match="Remote control is turned off"):
|
||||
await coordinator.async_send_command(bound, 'On')
|
||||
|
||||
Reference in New Issue
Block a user