Make Portainer container recreate service admin-only (#177301)

homeassistant/components/portainer/services.py

@@ -196,7 +196,8 @@ async def async_setup_services(hass: HomeAssistant) -> None:
         SERVICE_PRUNE_IMAGES_SCHEMA,
     )
 
-    hass.services.async_register(
+    service.async_register_admin_service(
+        hass,
         DOMAIN,
         SERVICE_RECREATE_CONTAINER,
         recreate_container,

tests/components/portainer/test_services.py

@@ -23,14 +23,14 @@ from homeassistant.components.portainer.services import (
     _async_get_device,
 )
 from homeassistant.const import ATTR_DEVICE_ID
-from homeassistant.core import HomeAssistant
+from homeassistant.core import Context, HomeAssistant
 from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
 from homeassistant.helpers.device_registry import DeviceRegistry
 
 from . import setup_integration
 from .conftest import TEST_CONTAINER_ID, TEST_CONTAINER_NAME, TEST_ENTRY
 
-from tests.common import MockConfigEntry
+from tests.common import MockConfigEntry, MockUser
 
 TEST_ENDPOINT_ID = 1
 TEST_DEVICE_IDENTIFIER = f"{TEST_ENTRY}_{TEST_ENDPOINT_ID}"
@@ -157,6 +157,33 @@ async def test_service_recreate_container(
     )
 
 
+async def test_service_recreate_container_non_admin_rejected(
+    hass: HomeAssistant,
+    device_registry: DeviceRegistry,
+    mock_portainer_client: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+    hass_read_only_user: MockUser,
+) -> None:
+    """Test recreate container service requires admin access."""
+
+    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, ATTR_PULL_IMAGE: True},
+        blocking=False,
+        context=Context(user_id=hass_read_only_user.id),
+    )
+    await hass.async_block_till_done()
+
+    mock_portainer_client.container_recreate.assert_not_called()
+
+
 @pytest.mark.parametrize(
     ("exception", "translation_key"),
     [