Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-12 11:38:47 +01:00
Code Activity

Add recreate services to Portainer (#167225)

Browse Source 
Co-authored-by: G Johansson <goran.johansson@shiftit.se>
This commit is contained in:
Erwin Douna
2026-05-21 22:52:07 +01:00
committed by GitHub
co-authored by G Johansson
parent 3187289913
commit 905b868c82
6 changed files with 274 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/portainer/icons.json
+3
View File
@@ -118,6 +118,9 @@
"services": {
"prune_images": {
"service": "mdi:delete-sweep"
},
"recreate_container": {
"service": "mdi:restart"
}
}
}
homeassistant/components/portainer/services.py
+94
View File
@@ -20,6 +20,9 @@ from .coordinator import PortainerConfigEntry
ATTR_DATE_UNTIL = "until"
ATTR_DANGLING = "dangling"
ATTR_TIMEOUT = "timeout"
ATTR_PULL_IMAGE = "pull_image"
ATTR_CONTAINER_DEVICE_ID = "container_device_id"
SERVICE_PRUNE_IMAGES = "prune_images"
SERVICE_PRUNE_IMAGES_SCHEMA = vol.Schema(
@@ -32,6 +35,17 @@ SERVICE_PRUNE_IMAGES_SCHEMA = vol.Schema(
},
)
SERVICE_RECREATE_CONTAINER = "recreate_container"
SERVICE_RECREATE_CONTAINER_SCHEMA = vol.Schema(
{
vol.Required(ATTR_CONTAINER_DEVICE_ID): cv.string,
vol.Optional(ATTR_TIMEOUT): vol.All(
cv.time_period, vol.Range(min=timedelta(minutes=1))
),
vol.Optional(ATTR_PULL_IMAGE): cv.boolean,
}
)
async def _extract_config_entry(service_call: ServiceCall) -> PortainerConfigEntry:
"""Extract config entry from the service call."""
@@ -75,6 +89,45 @@ async def _get_endpoint_id(
return endpoint_data.endpoint.id
async def _get_container_and_endpoint_ids(
call: ServiceCall,
) -> tuple[PortainerConfigEntry, int, str]:
"""Get config entry, endpoint ID and container ID from the container device ID."""
device_reg = dr.async_get(call.hass)
device = device_reg.async_get(call.data[ATTR_CONTAINER_DEVICE_ID])
if device is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_target",
)
config_entry: PortainerConfigEntry | None = None
for loaded_entry in call.hass.config_entries.async_loaded_entries(DOMAIN):
if loaded_entry.entry_id in device.config_entries:
config_entry = loaded_entry
break
if config_entry is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_target",
)
coordinator = config_entry.runtime_data
for data in coordinator.data.values():
for container_name, container_data in data.containers.items():
if (
DOMAIN,
f"{config_entry.entry_id}_{data.endpoint.id}_{container_name}",
) in device.identifiers:
return config_entry, data.endpoint.id, container_data.container.id
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_target",
)
async def prune_images(call: ServiceCall) -> None:
"""Prune unused images in Portainer, with more controls."""
config_entry = await _extract_config_entry(call)
@@ -104,6 +157,40 @@ async def prune_images(call: ServiceCall) -> None:
) from err
async def recreate_container(call: ServiceCall) -> None:
"""Recreate a container in Portainer, with more controls."""
config_entry, endpoint_id, container_id = await _get_container_and_endpoint_ids(
call
)
coordinator = config_entry.runtime_data
timeout: timedelta | None = call.data.get(ATTR_TIMEOUT)
try:
await coordinator.portainer.container_recreate(
endpoint_id=endpoint_id,
container_id=container_id,
**({"timeout": timeout} if timeout is not None else {}),
pull_image=call.data.get(ATTR_PULL_IMAGE, False),
)
except PortainerAuthenticationError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="invalid_auth_no_details",
) from err
except PortainerConnectionError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="cannot_connect_no_details",
) from err
except PortainerTimeoutError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="timeout_connect_no_details",
) from err
await coordinator.async_request_refresh()
async def async_setup_services(hass: HomeAssistant) -> None:
"""Set up services."""
@@ -113,3 +200,10 @@ async def async_setup_services(hass: HomeAssistant) -> None:
prune_images,
SERVICE_PRUNE_IMAGES_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_RECREATE_CONTAINER,
recreate_container,
SERVICE_RECREATE_CONTAINER_SCHEMA,
)
homeassistant/components/portainer/services.yaml
+17
View File
@@ -16,3 +16,20 @@ prune_images:
required: false
selector:
boolean: {}
recreate_container:
fields:
container_device_id:
required: true
selector:
device:
integration: portainer
model: Container
timeout:
required: false
selector:
duration:
pull_image:
required: false
selector:
boolean:
homeassistant/components/portainer/strings.json
+18
View File
@@ -235,6 +235,24 @@
}
},
"name": "Prune unused images"
},
"recreate_container": {
"description": "Recreates a container on a Portainer endpoint. This is more disruptive than a restart as the container will be stopped, removed, and then re-created with the same configuration. Use with caution.",
"fields": {
"container_device_id": {
"description": "The container to recreate.",
"name": "Container"
},
"pull_image": {
"description": "Whether to pull the image before recreating the container. This can be used to update the container to the latest version of the image.",
"name": "Pull image"
},
"timeout": {
"description": "The time to wait for the container to stop before killing it. If not provided, a default of 5 minutes will be used.",
"name": "Timeout"
}
},
"name": "Recreate container"
}
},
"system_health": {
tests/components/portainer/conftest.py
+3
View File
@@ -31,6 +31,8 @@ MOCK_TEST_CONFIG = {
TEST_ENTRY = "portainer_test_entry_123"
TEST_INSTANCE_ID = "299ab403-70a8-4c05-92f7-bf7a994d50df"
TEST_CONTAINER_NAME = "practical_morse"
TEST_CONTAINER_ID = "ee20facfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf"
@pytest.fixture
@@ -93,6 +95,7 @@ def mock_portainer_client() -> Generator[AsyncMock]:
client.stop_container = AsyncMock(return_value=None)
client.start_stack = AsyncMock(return_value=None)
client.stop_stack = AsyncMock(return_value=None)
client.container_recreate = AsyncMock(return_value=None)
yield client
tests/components/portainer/test_services.py
+139 -4
View File
@@ -13,9 +13,13 @@ from voluptuous import MultipleInvalid
from homeassistant.components.portainer.const import DOMAIN
from homeassistant.components.portainer.services import (
ATTR_CONTAINER_DEVICE_ID,
ATTR_DANGLING,
ATTR_DATE_UNTIL,
ATTR_PULL_IMAGE,
ATTR_TIMEOUT,
SERVICE_PRUNE_IMAGES,
SERVICE_RECREATE_CONTAINER,
)
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.core import HomeAssistant
@@ -23,13 +27,17 @@ from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.device_registry import DeviceRegistry
from . import setup_integration
from .conftest import TEST_ENTRY
from .conftest import TEST_CONTAINER_ID, TEST_CONTAINER_NAME, TEST_ENTRY
from tests.common import MockConfigEntry
TEST_ENDPOINT_ID = 1
TEST_DEVICE_IDENTIFIER = f"{TEST_ENTRY}_{TEST_ENDPOINT_ID}"
TEST_CONTAINER_DEVICE_IDENTIFIER = (
f"{TEST_ENTRY}_{TEST_ENDPOINT_ID}_{TEST_CONTAINER_NAME}"
)
async def test_services(
hass: HomeAssistant,
@@ -102,6 +110,99 @@ async def test_service_prune_images(
)
@pytest.mark.parametrize(
("call_arguments", "extra_expected_kwargs"),
[
({}, {"pull_image": False}),
(
{ATTR_TIMEOUT: timedelta(minutes=10)},
{"pull_image": False, "timeout": timedelta(minutes=10)},
),
(
{ATTR_TIMEOUT: timedelta(minutes=12), ATTR_PULL_IMAGE: True},
{"pull_image": True, "timeout": timedelta(minutes=12)},
),
],
ids=["no optional", "with duration", "with duration and pull_image"],
)
async def test_service_recreate_container(
hass: HomeAssistant,
device_registry: DeviceRegistry,
mock_portainer_client: AsyncMock,
mock_config_entry: MockConfigEntry,
call_arguments: dict,
extra_expected_kwargs: dict,
) -> None:
"""Test recreate container service with the variants."""
await setup_integration(hass, mock_config_entry)
container = device_registry.async_get_device(
identifiers={(DOMAIN, TEST_CONTAINER_DEVICE_IDENTIFIER)}
)
assert container is not None
await hass.services.async_call(
DOMAIN,
SERVICE_RECREATE_CONTAINER,
{
ATTR_CONTAINER_DEVICE_ID: container.id,
**call_arguments,
},
blocking=True,
)
mock_portainer_client.container_recreate.assert_called_once_with(
endpoint_id=TEST_ENDPOINT_ID,
container_id=TEST_CONTAINER_ID,
**extra_expected_kwargs,
)
@pytest.mark.parametrize(
("exception", "translation_key"),
[
(
PortainerAuthenticationError("auth"),
"invalid_auth_no_details",
),
(
PortainerConnectionError("conn"),
"cannot_connect_no_details",
),
(
PortainerTimeoutError("timeout"),
"timeout_connect_no_details",
),
],
)
async def test_service_recreate_container_portainer_exceptions(
hass: HomeAssistant,
device_registry: DeviceRegistry,
mock_portainer_client: AsyncMock,
mock_config_entry: MockConfigEntry,
exception: PortainerAuthenticationError
| PortainerConnectionError
| PortainerTimeoutError,
translation_key: str,
) -> None:
"""Test recreate container service handles Portainer exceptions."""
await setup_integration(hass, mock_config_entry)
container = device_registry.async_get_device(
identifiers={(DOMAIN, TEST_CONTAINER_DEVICE_IDENTIFIER)}
)
assert container is not None
mock_portainer_client.container_recreate.side_effect = exception
with pytest.raises(HomeAssistantError) as err:
await hass.services.async_call(
DOMAIN,
SERVICE_RECREATE_CONTAINER,
{ATTR_CONTAINER_DEVICE_ID: container.id},
blocking=True,
)
assert err.value.translation_key == translation_key
mock_portainer_client.container_recreate.assert_called_once()
async def test_service_validation_errors(
hass: HomeAssistant,
device_registry: DeviceRegistry,
@@ -115,8 +216,11 @@ async def test_service_validation_errors(
identifiers={(DOMAIN, TEST_DEVICE_IDENTIFIER)}
)
assert device is not None
container = device_registry.async_get_device(
identifiers={(DOMAIN, TEST_CONTAINER_DEVICE_IDENTIFIER)}
)
assert container is not None
# Test missing device_id
with pytest.raises(MultipleInvalid, match="required key not provided"):
await hass.services.async_call(
DOMAIN,
@@ -126,7 +230,6 @@ async def test_service_validation_errors(
)
mock_portainer_client.images_prune.assert_not_called()
# Test invalid until (too short, needs to be at least 1 minute)
with pytest.raises(MultipleInvalid, match="value must be at least"):
await hass.services.async_call(
DOMAIN,
@@ -136,7 +239,6 @@ async def test_service_validation_errors(
)
mock_portainer_client.images_prune.assert_not_called()
# Test invalid device
with pytest.raises(ServiceValidationError, match="Invalid device targeted"):
await hass.services.async_call(
DOMAIN,
@@ -146,6 +248,39 @@ async def test_service_validation_errors(
)
mock_portainer_client.images_prune.assert_not_called()
with pytest.raises(ServiceValidationError, match="Invalid device targeted"):
await hass.services.async_call(
DOMAIN,
SERVICE_RECREATE_CONTAINER,
{ATTR_CONTAINER_DEVICE_ID: "invalid_device_id"},
blocking=True,
)
mock_portainer_client.container_recreate.assert_not_called()
other_entry = MockConfigEntry(domain="well_no_portainer_for_sure")
other_entry.add_to_hass(hass)
non_portainer_device = device_registry.async_get_or_create(
config_entry_id=other_entry.entry_id,
identifiers={("well_no_portainer_for_sure", "some_identifier")},
)
with pytest.raises(ServiceValidationError, match="Invalid device targeted"):
await hass.services.async_call(
DOMAIN,
SERVICE_RECREATE_CONTAINER,
{ATTR_CONTAINER_DEVICE_ID: non_portainer_device.id},
blocking=True,
)
mock_portainer_client.container_recreate.assert_not_called()
with pytest.raises(ServiceValidationError, match="Invalid device targeted"):
await hass.services.async_call(
DOMAIN,
SERVICE_RECREATE_CONTAINER,
{ATTR_CONTAINER_DEVICE_ID: device.id},
blocking=True,
)
mock_portainer_client.container_recreate.assert_not_called()
@pytest.mark.parametrize(
("exception", "message"),