Compare commits

..

3 Commits

Author SHA1 Message Date
Paul Bottein
a21cb925af Clean code 2025-10-09 19:14:01 +02:00
Paul Bottein
3b2d8ddc1b Fix tests 2025-10-09 19:10:35 +02:00
Paul Bottein
496b6d4096 Let Home Assistant generate the entity id for esphome entities 2025-10-09 19:00:34 +02:00
9 changed files with 16 additions and 398 deletions

View File

@@ -90,7 +90,7 @@ def async_static_info_updated(
# Create new entity if it doesn't exist
if not old_info:
entity = entity_type(entry_data, platform.domain, info, state_type)
entity = entity_type(entry_data, info, state_type)
add_entities.append(entity)
continue
@@ -112,7 +112,7 @@ def async_static_info_updated(
old_info.device_id,
info.device_id,
)
entity = entity_type(entry_data, platform.domain, info, state_type)
entity = entity_type(entry_data, info, state_type)
add_entities.append(entity)
continue
@@ -162,7 +162,7 @@ def async_static_info_updated(
entry_data.async_signal_entity_removal(info_type, old_info.device_id, info.key)
# Create new entity with the new device_id
add_entities.append(entity_type(entry_data, platform.domain, info, state_type))
add_entities.append(entity_type(entry_data, info, state_type))
# Anything still in current_infos is now gone
if current_infos:
@@ -329,7 +329,6 @@ class EsphomeEntity(EsphomeBaseEntity, Generic[_InfoT, _StateT]):
def __init__(
self,
entry_data: RuntimeEntryData,
domain: str,
entity_info: EntityInfo,
state_type: type[_StateT],
) -> None:
@@ -343,7 +342,6 @@ class EsphomeEntity(EsphomeBaseEntity, Generic[_InfoT, _StateT]):
self._state_type = state_type
self._on_static_info_update(entity_info)
device_name = device_info.name
# Determine the device connection based on whether this entity belongs to a sub device
if entity_info.device_id:
# Entity belongs to a sub device
@@ -352,27 +350,12 @@ class EsphomeEntity(EsphomeBaseEntity, Generic[_InfoT, _StateT]):
(DOMAIN, f"{device_info.mac_address}_{entity_info.device_id}")
}
)
# Use the pre-computed device_id_to_name mapping for O(1) lookup
device_name = entry_data.device_id_to_name.get(
entity_info.device_id, device_info.name
)
else:
# Entity belongs to the main device
self._attr_device_info = DeviceInfo(
connections={(dr.CONNECTION_NETWORK_MAC, device_info.mac_address)}
)
if entity_info.name:
self.entity_id = f"{domain}.{device_name}_{entity_info.name}"
else:
# https://github.com/home-assistant/core/issues/132532
# If name is not set, ESPHome will use the sanitized friendly name
# as the name, however we want to use the original object_id
# as the entity_id before it is sanitized since the sanitizer
# is not utf-8 aware. In this case, its always going to be
# an empty string so we drop the object_id.
self.entity_id = f"{domain}.{device_name}"
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
entry_data = self._entry_data

View File

@@ -18,8 +18,7 @@ from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .coordinator import PortainerCoordinator
_PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR, Platform.SWITCH]
_PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.BUTTON, Platform.SWITCH]
type PortainerConfigEntry = ConfigEntry[PortainerCoordinator]

View File

@@ -1,10 +1,5 @@
{
"entity": {
"sensor": {
"image": {
"default": "mdi:docker"
}
},
"switch": {
"container": {
"default": "mdi:arrow-down-box",

View File

@@ -1,83 +0,0 @@
"""Sensor platform for Portainer integration."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from pyportainer.models.docker import DockerContainer
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import PortainerConfigEntry, PortainerCoordinator
from .entity import PortainerContainerEntity, PortainerCoordinatorData
@dataclass(frozen=True, kw_only=True)
class PortainerSensorEntityDescription(SensorEntityDescription):
"""Class to hold Portainer sensor description."""
value_fn: Callable[[DockerContainer], str | None]
CONTAINER_SENSORS: tuple[PortainerSensorEntityDescription, ...] = (
PortainerSensorEntityDescription(
key="image",
translation_key="image",
value_fn=lambda data: data.image,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: PortainerConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Portainer sensors based on a config entry."""
coordinator = entry.runtime_data
async_add_entities(
PortainerContainerSensor(
coordinator,
entity_description,
container,
endpoint,
)
for endpoint in coordinator.data.values()
for container in endpoint.containers.values()
for entity_description in CONTAINER_SENSORS
)
class PortainerContainerSensor(PortainerContainerEntity, SensorEntity):
"""Representation of a Portainer container sensor."""
entity_description: PortainerSensorEntityDescription
def __init__(
self,
coordinator: PortainerCoordinator,
entity_description: PortainerSensorEntityDescription,
device_info: DockerContainer,
via_device: PortainerCoordinatorData,
) -> None:
"""Initialize the Portainer container sensor."""
self.entity_description = entity_description
super().__init__(device_info, coordinator, via_device)
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{self.device_name}_{entity_description.key}"
@property
def available(self) -> bool:
"""Return if the device is available."""
return super().available and self.endpoint_id in self.coordinator.data
@property
def native_value(self) -> str | None:
"""Return the state of the sensor."""
return self.entity_description.value_fn(
self.coordinator.data[self.endpoint_id].containers[self.device_id]
)

View File

@@ -46,11 +46,6 @@
"name": "Status"
}
},
"sensor": {
"image": {
"name": "Image"
}
},
"switch": {
"container": {
"name": "Container"

View File

@@ -260,11 +260,11 @@ class TriggerConfig:
class TriggerActionType(Protocol):
"""Protocol type for trigger action callback."""
def __call__(
async def __call__(
self,
run_variables: dict[str, Any],
context: Context | None = None,
) -> Coroutine[Any, Any, Any] | Any:
) -> Any:
"""Define action callback type."""
@@ -444,8 +444,8 @@ async def async_validate_trigger_config(
def _trigger_action_wrapper(
hass: HomeAssistant, action: TriggerActionType, conf: ConfigType
) -> TriggerActionType:
hass: HomeAssistant, action: Callable, conf: ConfigType
) -> Callable:
"""Wrap trigger action with extra vars if configured.
If action is a coroutine function, a coroutine function will be returned.
@@ -477,7 +477,7 @@ def _trigger_action_wrapper(
else:
@functools.wraps(action)
def with_vars(
async def with_vars(
run_variables: dict[str, Any], context: Context | None = None
) -> Any:
"""Wrap action with extra vars."""

View File

@@ -489,7 +489,7 @@ async def test_entity_without_name_device_with_friendly_name(
states=states,
device_info={"friendly_name": "The Best Mixer", "name": "mixer"},
)
state = hass.states.get("binary_sensor.mixer")
state = hass.states.get("binary_sensor.the_best_mixer")
assert state is not None
assert state.state == STATE_ON
# Make sure we have set the name to `None` as otherwise
@@ -601,7 +601,7 @@ async def test_entity_id_preserved_on_upgrade_when_in_storage(
states=states,
device_info={"friendly_name": "The Best Mixer", "name": "mixer"},
)
state = hass.states.get("binary_sensor.mixer_my")
state = hass.states.get("binary_sensor.the_best_mixer_my")
assert state is not None
# now rename the entity
ent_reg_entry = entity_registry.async_get_or_create(
@@ -874,8 +874,8 @@ async def test_entity_friendly_names_with_empty_device_names(
)
# Check entity friendly name on sub-device with empty name
# Since sub device has empty name, it falls back to main device name "test"
state_1 = hass.states.get("binary_sensor.test_motion_detected")
# Since sub device has empty name, it falls back to main device name "Main device"
state_1 = hass.states.get("binary_sensor.main_device_motion_detected")
assert state_1 is not None
# With has_entity_name, friendly name is "{device_name} {entity_name}"
# Since sub-device falls back to main device name: "Main Device Motion Detected"
@@ -894,7 +894,7 @@ async def test_entity_friendly_names_with_empty_device_names(
assert state_3.attributes[ATTR_FRIENDLY_NAME] == "Kitchen Light"
# Test entity on main device
state_4 = hass.states.get("binary_sensor.test_main_status")
state_4 = hass.states.get("binary_sensor.main_device_main_status")
assert state_4 is not None
assert state_4.attributes[ATTR_FRIENDLY_NAME] == "Main Device Main Status"
@@ -1050,6 +1050,7 @@ async def test_entity_id_uses_sub_device_name(
device_info = {
"devices": sub_devices,
"name": "main_device",
"friendly_name": None,
}
# Create entities that belong to different devices
@@ -1131,6 +1132,7 @@ async def test_entity_id_with_empty_sub_device_name(
device_info = {
"devices": sub_devices,
"name": "main_device",
"friendly_name": None,
}
# Create entity on sub device with empty name

View File

@@ -1,241 +0,0 @@
# serializer version: 1
# name: test_all_entities[sensor.focused_einstein_image-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.focused_einstein_image',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Image',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'image',
'unique_id': 'portainer_test_entry_123_focused_einstein_image',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.focused_einstein_image-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'focused_einstein Image',
}),
'context': <ANY>,
'entity_id': 'sensor.focused_einstein_image',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'docker.io/library/redis:7',
})
# ---
# name: test_all_entities[sensor.funny_chatelet_image-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.funny_chatelet_image',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Image',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'image',
'unique_id': 'portainer_test_entry_123_funny_chatelet_image',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.funny_chatelet_image-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'funny_chatelet Image',
}),
'context': <ANY>,
'entity_id': 'sensor.funny_chatelet_image',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'docker.io/library/ubuntu:latest',
})
# ---
# name: test_all_entities[sensor.practical_morse_image-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.practical_morse_image',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Image',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'image',
'unique_id': 'portainer_test_entry_123_practical_morse_image',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.practical_morse_image-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'practical_morse Image',
}),
'context': <ANY>,
'entity_id': 'sensor.practical_morse_image',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'docker.io/library/python:3.13-slim',
})
# ---
# name: test_all_entities[sensor.serene_banach_image-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.serene_banach_image',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Image',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'image',
'unique_id': 'portainer_test_entry_123_serene_banach_image',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.serene_banach_image-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'serene_banach Image',
}),
'context': <ANY>,
'entity_id': 'sensor.serene_banach_image',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'docker.io/library/nginx:latest',
})
# ---
# name: test_all_entities[sensor.stoic_turing_image-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.stoic_turing_image',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Image',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'image',
'unique_id': 'portainer_test_entry_123_stoic_turing_image',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.stoic_turing_image-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'stoic_turing Image',
}),
'context': <ANY>,
'entity_id': 'sensor.stoic_turing_image',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'docker.io/library/postgres:15',
})
# ---

View File

@@ -1,32 +0,0 @@
"""Tests for the Portainer sensor platform."""
from unittest.mock import patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
@pytest.mark.usefixtures("mock_portainer_client")
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test all entities."""
with patch(
"homeassistant.components.portainer._PLATFORMS",
[Platform.SENSOR],
):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(
hass, entity_registry, snapshot, mock_config_entry.entry_id
)