Write protect entity options (#90185)

This commit is contained in:
Erik Montnemery 2023-03-28 22:56:51 +02:00 committed by GitHub
parent 0550b17d54
commit e22618a555
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 41 additions and 8 deletions

View File

@ -1,6 +1,7 @@
"""Preference management for camera component.""" """Preference management for camera component."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Mapping
from dataclasses import asdict, dataclass from dataclasses import asdict, dataclass
from typing import Final, cast from typing import Final, cast
@ -89,7 +90,7 @@ class CameraPreferences:
# Get preload stream setting from prefs # Get preload stream setting from prefs
# Get orientation setting from entity registry # Get orientation setting from entity registry
reg_entry = er.async_get(self._hass).async_get(entity_id) reg_entry = er.async_get(self._hass).async_get(entity_id)
er_prefs = reg_entry.options.get(DOMAIN, {}) if reg_entry else {} er_prefs: Mapping = reg_entry.options.get(DOMAIN, {}) if reg_entry else {}
preload_prefs = await self._store.async_load() or {} preload_prefs = await self._store.async_load() or {}
settings = DynamicStreamSettings( settings = DynamicStreamSettings(
preload_stream=cast( preload_stream=cast(

View File

@ -737,7 +737,7 @@ class SensorEntity(Entity):
or "suggested_display_precision" not in self.registry_entry.options or "suggested_display_precision" not in self.registry_entry.options
): ):
return return
sensor_options = self.registry_entry.options.get(DOMAIN, {}) sensor_options: Mapping[str, Any] = self.registry_entry.options.get(DOMAIN, {})
if ( if (
"suggested_display_precision" in sensor_options "suggested_display_precision" in sensor_options
and sensor_options["suggested_display_precision"] == display_precision and sensor_options["suggested_display_precision"] == display_precision

View File

@ -12,6 +12,7 @@ from __future__ import annotations
from collections import UserDict from collections import UserDict
from collections.abc import Callable, Iterable, Mapping, ValuesView from collections.abc import Callable, Iterable, Mapping, ValuesView
import logging import logging
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, TypeVar, cast from typing import TYPE_CHECKING, Any, TypeVar, cast
import attr import attr
@ -111,6 +112,29 @@ DISLAY_DICT_OPTIONAL = (
) )
class _EntityOptions(UserDict[str, MappingProxyType]):
"""Container for entity options."""
def __init__(self, data: Mapping[str, Mapping] | None) -> None:
"""Initialize."""
super().__init__()
if data is None:
return
self.data = {key: MappingProxyType(val) for key, val in data.items()}
def __setitem__(self, key: str, entry: Mapping) -> None:
"""Add an item."""
raise NotImplementedError
def __delitem__(self, key: str) -> None:
"""Remove an item."""
raise NotImplementedError
def as_dict(self) -> dict[str, dict]:
"""Return dictionary version."""
return {key: dict(val) for key, val in self.data.items()}
@attr.s(slots=True, frozen=True) @attr.s(slots=True, frozen=True)
class RegistryEntry: class RegistryEntry:
"""Entity Registry Entry.""" """Entity Registry Entry."""
@ -132,10 +156,7 @@ class RegistryEntry:
id: str = attr.ib(factory=uuid_util.random_uuid_hex) id: str = attr.ib(factory=uuid_util.random_uuid_hex)
has_entity_name: bool = attr.ib(default=False) has_entity_name: bool = attr.ib(default=False)
name: str | None = attr.ib(default=None) name: str | None = attr.ib(default=None)
options: EntityOptionsType = attr.ib( options: _EntityOptions = attr.ib(default=None, converter=_EntityOptions)
default=None,
converter=attr.converters.default_if_none(factory=dict), # type: ignore[misc]
)
# As set by integration # As set by integration
original_device_class: str | None = attr.ib(default=None) original_device_class: str | None = attr.ib(default=None)
original_icon: str | None = attr.ib(default=None) original_icon: str | None = attr.ib(default=None)
@ -930,7 +951,7 @@ class EntityRegistry:
If the domain options are set to None, they will be removed. If the domain options are set to None, they will be removed.
""" """
old = self.entities[entity_id] old = self.entities[entity_id]
new_options = { new_options: dict[str, Mapping] = {
key: value for key, value in old.options.items() if key != domain key: value for key, value in old.options.items() if key != domain
} }
if options is not None: if options is not None:
@ -1010,7 +1031,7 @@ class EntityRegistry:
"id": entry.id, "id": entry.id,
"has_entity_name": entry.has_entity_name, "has_entity_name": entry.has_entity_name,
"name": entry.name, "name": entry.name,
"options": entry.options, "options": entry.options.as_dict(),
"original_device_class": entry.original_device_class, "original_device_class": entry.original_device_class,
"original_icon": entry.original_icon, "original_icon": entry.original_icon,
"original_name": entry.original_name, "original_name": entry.original_name,

View File

@ -747,6 +747,16 @@ async def test_update_entity_options(entity_registry: er.EntityRegistry) -> None
assert entry.options == {} assert entry.options == {}
assert new_entry_1.options == {"light": {"minimum_brightness": 20}} assert new_entry_1.options == {"light": {"minimum_brightness": 20}}
# Test it's not possible to modify the options
with pytest.raises(NotImplementedError):
new_entry_1.options["blah"] = {}
with pytest.raises(NotImplementedError):
new_entry_1.options["light"] = {}
with pytest.raises(TypeError):
new_entry_1.options["light"]["blah"] = 123
with pytest.raises(TypeError):
new_entry_1.options["light"]["minimum_brightness"] = 123
entity_registry.async_update_entity_options( entity_registry.async_update_entity_options(
entry.entity_id, "light", {"minimum_brightness": 30} entry.entity_id, "light", {"minimum_brightness": 30}
) )

View File

@ -170,6 +170,7 @@ class HomeAssistantSnapshotSerializer(AmberDataSerializer):
"config_entry_id": ANY, "config_entry_id": ANY,
"device_id": ANY, "device_id": ANY,
"id": ANY, "id": ANY,
"options": data.options.as_dict(),
} }
) )
serialized.pop("_partial_repr") serialized.pop("_partial_repr")