starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 01:11:51 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix Portainer image update entity turning unknown after a container recreate (#182648)

Browse Source
This commit is contained in:
Franck Nijhof
2026-09-22 08:28:19 +02:00
committed by GitHub
parent 90233bf798
commit 0634fd9f17
2 changed files with 217 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/portainer/coordinator.py
+46 -13
View File
@@ -16,6 +16,7 @@ from pyportainer import (
Portainer,
PortainerAuthenticationError,
PortainerConnectionError,
PortainerError,
PortainerEventListener,
PortainerEventListenerResult,
PortainerTimeoutError,
@@ -217,6 +218,9 @@ class PortainerCoordinator(
self._image_cache: dict[
tuple[int, str], tuple[float, LocalImageInformation]
] = {}
self._image_status_cache: dict[
tuple[int, str], tuple[float, PortainerImageUpdateStatus]
] = {}
self._event_listeners: dict[int, PortainerEventListener] = {}
self._event_listeners_enabled = False
self._container_ids_by_endpoint: dict[int, dict[str, str]] = {}
@@ -341,19 +345,7 @@ class PortainerCoordinator(
container_inspect = container_inspects[container_name]
local_image = local_images[container_name]
image_status = (
(
result.status
if (
result := self.watcher.results.get(
(endpoint.id, container.id)
)
)
else None
)
if self.watcher
else None
)
image_status = await self._get_image_status(endpoint.id, container)
# Check if container belongs to a stack via docker compose label
stack_name: str | None = (
@@ -618,6 +610,47 @@ class PortainerCoordinator(
)
return local_image
async def _get_image_status(
self, endpoint_id: int, container: DockerContainer
) -> PortainerImageUpdateStatus | None:
"""Return the image update status, checking containers the watcher has not seen."""
if self.watcher is None:
return None
if result := self.watcher.results.get((endpoint_id, container.id)):
return result.status
# A recreated container gets a new ID, which the watcher only picks up on
# its next run. Check its image now instead of reporting unknown until then.
if (
self.watcher.last_check is None
or container.state != DockerContainerState.RUNNING
or not container.image
):
return None
cache_key = (endpoint_id, container.image)
if cached := self._image_status_cache.get(cache_key):
cached_at, image_status = cached
if cached_at >= self.watcher.last_check:
return image_status
try:
image_status = await self.portainer.container_image_status(
endpoint_id, container.image
)
except PortainerError as err:
_LOGGER.debug(
"Failed to check image %s on endpoint %d: %s",
container.image,
endpoint_id,
err,
)
return None
self._image_status_cache[cache_key] = (time.time(), image_status)
return image_status
def _async_sync_event_listeners(
self, mapped_endpoints: dict[int, PortainerCoordinatorData]
) -> None:
tests/components/portainer/test_update.py
+171 -3
View File
@@ -1,25 +1,44 @@
"""Tests for the Portainer update platform."""
from typing import Any, cast
from unittest.mock import AsyncMock, MagicMock, patch
from freezegun.api import FrozenDateTimeFactory
from pyportainer.exceptions import (
PortainerAuthenticationError,
PortainerConnectionError,
)
from pyportainer.models.docker import DockerContainer, PortainerImageUpdateStatus
from pyportainer.watcher import PortainerImageWatcherResult
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.update import ATTR_INSTALLED_VERSION
from homeassistant.const import Platform
from homeassistant.components.portainer.const import DOMAIN
from homeassistant.components.portainer.coordinator import DEFAULT_SCAN_INTERVAL
from homeassistant.components.update import ATTR_INSTALLED_VERSION, ATTR_LATEST_VERSION
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
async_load_json_array_fixture,
snapshot_platform,
)
ENTITY_ID = "update.funny_chatelet_image_update_available"
CONTAINER_IMAGE = "docker.io/library/ubuntu:latest"
INSTALLED_DIGEST = (
"sha256:afcc7f1ac1b49db317a7196c902e61c6c3c4607d63599ee1a82d702d249a0ccb"
)
RECREATED_CONTAINER_ID = (
"0011facfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf"
)
@pytest.fixture(autouse=True)
@@ -158,3 +177,152 @@ async def test_update_using_cache(
)
mock_portainer_client.get_image.assert_not_called()
async def _watch_all_containers(hass: HomeAssistant, watcher: MagicMock) -> None:
"""Give the watcher a result for every container, as after its first run."""
containers = cast(
list[dict[str, Any]],
await async_load_json_array_fixture(hass, "containers.json", DOMAIN),
)
watcher.results = {
(1, container["Id"]): PortainerImageWatcherResult(
endpoint_id=1,
container_id=container["Id"],
status=PortainerImageUpdateStatus(
update_available=True,
local_digest=INSTALLED_DIGEST,
registry_digest="sha256:newdigest123456789",
),
)
for container in containers
}
watcher.last_check = 1234
async def _recreate_container(hass: HomeAssistant, client: AsyncMock) -> None:
"""Give the funny_chatelet container a new ID, as a recreate does."""
containers = cast(
list[dict[str, Any]],
await async_load_json_array_fixture(hass, "containers.json", DOMAIN),
)
recreated = next(
container for container in containers if "/funny_chatelet" in container["Names"]
)
recreated["Id"] = RECREATED_CONTAINER_ID
client.get_containers.return_value = [
DockerContainer.from_dict(container) for container in containers
]
async def test_update_recreated_container(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_portainer_client: AsyncMock,
mock_portainer_watcher: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test a recreated container gets its image checked instead of staying unknown."""
await _watch_all_containers(hass, mock_portainer_watcher)
with patch(
"homeassistant.components.portainer._PLATFORMS",
[Platform.UPDATE],
):
await setup_integration(hass, mock_config_entry)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_ON
# The watcher only has a result for the old container ID
await _recreate_container(hass, mock_portainer_client)
mock_portainer_client.container_image_status.return_value = (
PortainerImageUpdateStatus(
update_available=False,
local_digest=INSTALLED_DIGEST,
registry_digest=INSTALLED_DIGEST,
)
)
freezer.tick(DEFAULT_SCAN_INTERVAL)
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done(wait_background_tasks=True)
mock_portainer_client.container_image_status.assert_called_once_with(
1, CONTAINER_IMAGE
)
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
assert state.attributes[ATTR_LATEST_VERSION] == "sha256:afcc7f1ac1b4"
# The result is kept until the watcher runs again, not fetched every poll
freezer.tick(DEFAULT_SCAN_INTERVAL)
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done(wait_background_tasks=True)
mock_portainer_client.container_image_status.assert_called_once()
async def test_update_recreated_container_before_watcher_ran(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_portainer_client: AsyncMock,
mock_portainer_watcher: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test a container is left to the watcher's first run."""
with patch(
"homeassistant.components.portainer._PLATFORMS",
[Platform.UPDATE],
):
await setup_integration(hass, mock_config_entry)
await _recreate_container(hass, mock_portainer_client)
freezer.tick(DEFAULT_SCAN_INTERVAL)
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done(wait_background_tasks=True)
mock_portainer_client.container_image_status.assert_not_called()
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_UNKNOWN
@pytest.mark.parametrize(
"exception",
[
pytest.param(PortainerConnectionError("conn"), id="connection"),
pytest.param(PortainerAuthenticationError("auth"), id="authentication"),
],
)
async def test_update_recreated_container_check_fails(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_portainer_client: AsyncMock,
mock_portainer_watcher: MagicMock,
mock_config_entry: MockConfigEntry,
exception: Exception,
) -> None:
"""Test a failed image check leaves the update unknown without failing the refresh."""
await _watch_all_containers(hass, mock_portainer_watcher)
with patch(
"homeassistant.components.portainer._PLATFORMS",
[Platform.UPDATE],
):
await setup_integration(hass, mock_config_entry)
await _recreate_container(hass, mock_portainer_client)
mock_portainer_client.container_image_status.side_effect = exception
freezer.tick(DEFAULT_SCAN_INTERVAL)
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done(wait_background_tasks=True)
assert mock_config_entry.runtime_data.last_update_success
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == STATE_UNKNOWN