Add reconfigure flow to collection_image (#182108)

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

homeassistant/components/collection_image/config_flow.py

@@ -8,6 +8,8 @@ from homeassistant.components.image import DOMAIN as IMAGE_DOMAIN
 from homeassistant.components.media_player import BrowseError, MediaClass
 from homeassistant.components.media_source import URI_SCHEME, async_browse_media
 from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.selector import MediaSelector
 
 from .const import CONF_MEDIA, DOMAIN
@@ -23,57 +25,102 @@ STEP_USER_DATA_SCHEMA = probatio.Schema(
 )
 
 
+async def _async_validate_media(
+    hass: HomeAssistant,
+    user_input: dict[str, Any],
+) -> tuple[str | None, dict[str, str], dict[str, str]]:
+    """Validate selected directories and return title and form errors."""
+    errors: dict[str, str] = {}
+    placeholders: dict[str, str] = {}
+    found_pictures = False
+    title = "Unnamed collection"
+
+    for user_media in user_input[CONF_MEDIA]:
+        if user_media["media_content_id"] == IMAGE_MEDIA_URI:
+            errors[CONF_MEDIA] = "invalid_selection"
+            placeholders["error"] = IMAGE_MEDIA_URI
+            break
+
+        try:
+            browse = await async_browse_media(
+                hass,
+                user_media["media_content_id"],
+            )
+        except BrowseError as err:
+            errors[CONF_MEDIA] = "failed_browse"
+            placeholders["error"] = str(err)
+            break
+
+        if (
+            not found_pictures
+            and browse.children
+            and any(item.media_class == MediaClass.IMAGE for item in browse.children)
+        ):
+            found_pictures = True
+            if browse.title:
+                title = f"{browse.title} collection"
+
+    if not errors and not found_pictures:
+        errors[CONF_MEDIA] = "selected_media_no_images"
+
+    return (title if not errors else None), errors, placeholders
+
+
 class CollectionImageConfigFlow(ConfigFlow, domain=DOMAIN):
     """Handle a config flow for Collection Image."""
 
-    @override
-    async def async_step_user(
+    async def async_step_reconfigure(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
-        """Handle the initial step."""
+        """Handle reconfiguration."""
         errors: dict[str, str] = {}
         placeholders: dict[str, str] = {}
-        found_pictures = False
-        title = "Unnamed collection"
+        entry = self._get_reconfigure_entry()
         if user_input is not None:
-            user_media_list = user_input[CONF_MEDIA]
-            for user_media in user_media_list:
-                if user_media["media_content_id"] == IMAGE_MEDIA_URI:
-                    errors["media"] = "invalid_selection"
-                    placeholders["error"] = IMAGE_MEDIA_URI
-                    break
-                try:
-                    browse = await async_browse_media(
-                        self.hass, user_media["media_content_id"]
-                    )
-                except BrowseError as err:
-                    errors["media"] = "failed_browse"
-                    placeholders["error"] = str(err)
-                    break
-                else:
-                    if (
-                        not found_pictures
-                        and browse.children
-                        and any(
-                            item.media_class == MediaClass.IMAGE
-                            for item in browse.children
-                        )
-                    ):
-                        found_pictures = True
-                        if browse.title:
-                            title = f"{browse.title} collection"
-            if "media" not in errors:
-                if found_pictures:
-                    return self.async_create_entry(
-                        title=title,
-                        data=user_input,
-                    )
-                errors["media"] = "selected_media_no_images"
+            title, errors, placeholders = await _async_validate_media(
+                self.hass,
+                user_input,
+            )
+            if title is not None:
+                return self.async_update_reload_and_abort(
+                    entry, data_updates=user_input
+                )
 
         return self.async_show_form(
-            step_id="user",
+            step_id="reconfigure",
             data_schema=self.add_suggested_values_to_schema(
-                STEP_USER_DATA_SCHEMA, user_input
+                STEP_USER_DATA_SCHEMA,
+                user_input or {CONF_MEDIA: cv.ensure_list(entry.data[CONF_MEDIA])},
+            ),
+            errors=errors,
+            description_placeholders=placeholders,
+        )
+
+    @override
+    async def async_step_user(
+        self,
+        user_input: dict[str, Any] | None = None,
+    ) -> ConfigFlowResult:
+        """Handle initial setup."""
+        errors: dict[str, str] = {}
+        placeholders: dict[str, str] = {}
+
+        if user_input is not None:
+            title, errors, placeholders = await _async_validate_media(
+                self.hass,
+                user_input,
+            )
+            if title is not None:
+                return self.async_create_entry(
+                    title=title,
+                    data=user_input,
+                )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=self.add_suggested_values_to_schema(
+                STEP_USER_DATA_SCHEMA,
+                user_input,
             ),
             errors=errors,
             description_placeholders=placeholders,

homeassistant/components/collection_image/quality_scale.yaml

@@ -92,9 +92,7 @@ rules:
   icon-translations:
     status: exempt
     comment: No meaningful icon translations for an image entity.
-  reconfiguration-flow:
-    status: exempt
-    comment: Nothing to reconfigure.
+  reconfiguration-flow: done
   repair-issues:
     status: exempt
     comment: Nothing to repair.

homeassistant/components/collection_image/strings.json

@@ -6,6 +6,15 @@
       "selected_media_no_images": "The selected media has no images. Please select a media directory with images."
     },
     "step": {
+      "reconfigure": {
+        "data": {
+          "media": "[%key:component::collection_image::config::step::user::data::media%]"
+        },
+        "data_description": {
+          "media": "[%key:component::collection_image::config::step::user::data_description::media%]"
+        },
+        "description": "Updates the source media used for the collection."
+      },
       "user": {
         "data": {
           "media": "Media"

tests/components/collection_image/helpers.py

@@ -1,29 +1,37 @@
 """Helper utilities for collection image tests."""
 
-from homeassistant.components.collection_image.const import DOMAIN
+from typing import Any
+
+from homeassistant.components.collection_image.const import CONF_MEDIA, DOMAIN
 from homeassistant.components.media_player import BrowseMedia, MediaClass
 from homeassistant.components.media_source import BrowseMediaSource
 
 from tests.common import MockConfigEntry
 
 
-def config_entry_from_uri(uri: str | list[str]) -> MockConfigEntry:
-    """Construct a mock config entry from one URI or a list of URIs."""
+def data_from_uri(uri: str | list[str]) -> dict[str, Any]:
+    """Construct a data entry from one URI or a list of URIs."""
 
-    def media_item(content_id: str) -> dict[str, str]:
+    def media_item(content_id: str) -> dict[str, Any]:
         return {
             "media_content_id": content_id,
             "media_content_type": "",
+            "metadata": {"a": "b"},
         }
 
-    media: dict[str, str] | list[dict[str, str]]
+    media: dict[str, Any] | list[dict[str, Any]]
     if isinstance(uri, str):
         media = media_item(uri)
     else:
         media = [media_item(item) for item in uri]
 
+    return {CONF_MEDIA: media}
+
+
+def config_entry_from_uri(uri: str | list[str]) -> MockConfigEntry:
+    """Construct a mock config entry from one URI or a list of URIs."""
     return MockConfigEntry(
-        data={"media": media},
+        data=data_from_uri(uri),
         domain=DOMAIN,
         title="Random Image",
     )

tests/components/collection_image/test_config_flow.py

@@ -1,14 +1,15 @@
 """Test the Collection Image config flow."""
 
-from unittest.mock import AsyncMock, patch
-
+from freezegun import freeze_time
+from freezegun.api import FrozenDateTimeFactory
 import pytest
 
 from homeassistant import config_entries
 from homeassistant.components.collection_image.config_flow import IMAGE_MEDIA_URI
-from homeassistant.components.collection_image.const import DOMAIN
+from homeassistant.components.collection_image.const import CONF_MEDIA, DOMAIN
 from homeassistant.core import HomeAssistant
 from homeassistant.data_entry_flow import FlowResultType
+from homeassistant.util import slugify
 
 from .const import (
     MOCK_MEDIA_DIR_URI_1,
@@ -16,30 +17,12 @@ from .const import (
     MOCK_MEDIA_DIR_URI_BROWSE_ERROR,
     MOCK_MEDIA_DIR_URI_EMPTY,
 )
+from .helpers import data_from_uri
 
+from tests.common import MockConfigEntry
 
-@pytest.fixture
-def mock_setup_entry():
-    """Mock collection_image setup successfully."""
-
-    with patch(
-        "homeassistant.components.collection_image.async_setup_entry",
-        new=AsyncMock(return_value=True),
-    ) as mock_setup:
-        yield mock_setup
-
-
-def _data_from_uris(uris: list[str]) -> dict:
-    return {
-        "media": [
-            {
-                "media_content_id": uri,
-                "media_content_type": "",
-                "metadata": {"a": "b"},
-            }
-            for uri in uris
-        ]
-    }
+TEST_TIME = "2026-09-12T07:12:00+00:00"
+TEST_TIME_NEXT = "2026-09-12T07:30:00+00:00"
 
 
 @pytest.mark.parametrize(
@@ -51,8 +34,9 @@ def _data_from_uris(uris: list[str]) -> dict:
     ],
 )
 @pytest.mark.usefixtures("mock_media_source")
+@freeze_time(TEST_TIME)
 async def test_config_flow(
-    hass: HomeAssistant, mock_setup_entry, uris: list[str], expected_title: str
+    hass: HomeAssistant, uris: list[str], expected_title: str
 ) -> None:
     """Test the config flow."""
 
@@ -62,14 +46,18 @@ async def test_config_flow(
     assert result.get("type") is FlowResultType.FORM
     assert result.get("errors") == {}
 
-    data = _data_from_uris(uris)
+    data = data_from_uri(uris)
 
     result = await hass.config_entries.flow.async_configure(result["flow_id"], data)
 
     assert result.get("type") is FlowResultType.CREATE_ENTRY
     assert result.get("title") == expected_title
     assert result.get("data") == data
-    assert len(mock_setup_entry.mock_calls) == 1
+
+    await hass.async_block_till_done()
+
+    state = hass.states.get(f"image.{slugify(expected_title)}")
+    assert state and state.state == TEST_TIME
 
 
 @pytest.mark.parametrize(
@@ -106,10 +94,10 @@ async def test_config_flow(
         ),
     ],
 )
+@freeze_time(TEST_TIME)
 @pytest.mark.usefixtures("mock_media_source")
 async def test_config_flow_error(
     hass: HomeAssistant,
-    mock_setup_entry,
     uris: list[str],
     error: str,
     placeholders: dict,
@@ -122,7 +110,7 @@ async def test_config_flow_error(
     assert result.get("type") is FlowResultType.FORM
     assert result.get("errors") == {}
 
-    data = _data_from_uris(uris)
+    data = data_from_uri(uris)
     result = await hass.config_entries.flow.async_configure(result["flow_id"], data)
     await hass.async_block_till_done()
 
@@ -133,21 +121,20 @@ async def test_config_flow_error(
     media_key = next(
         key
         for key in result["data_schema"].schema
-        if getattr(key, "schema", key) == "media"
+        if getattr(key, "schema", key) == CONF_MEDIA
     )
     for idx, uri in enumerate(uris):
         assert media_key.description["suggested_value"][idx]["media_content_id"] == uri
         assert (
             media_key.description["suggested_value"][idx]["metadata"]
-            == data["media"][idx]["metadata"]
+            == data[CONF_MEDIA][idx]["metadata"]
         )
 
-    assert result.get("errors") == {"media": error}
+    assert result.get("errors") == {CONF_MEDIA: error}
     assert result.get("description_placeholders") == placeholders
-    assert len(mock_setup_entry.mock_calls) == 0
 
     # Try again successfully to ensure we can recover from errors
-    data = _data_from_uris([MOCK_MEDIA_DIR_URI_1])
+    data = data_from_uri([MOCK_MEDIA_DIR_URI_1])
     expected_title = "My pictures collection"
 
     result = await hass.config_entries.flow.async_configure(result["flow_id"], data)
@@ -155,4 +142,93 @@ async def test_config_flow_error(
     assert result.get("type") is FlowResultType.CREATE_ENTRY
     assert result.get("title") == expected_title
     assert result.get("data") == data
-    assert len(mock_setup_entry.mock_calls) == 1
+
+    await hass.async_block_till_done()
+
+    state = hass.states.get(f"image.{slugify(expected_title)}")
+    assert state and state.state == TEST_TIME
+
+
+@pytest.mark.parametrize(
+    ("entry_data", "expected_uri"),
+    [
+        pytest.param(
+            data_from_uri([MOCK_MEDIA_DIR_URI_1]),
+            MOCK_MEDIA_DIR_URI_1,
+            id="legacy-data-array",
+        ),
+        pytest.param(
+            data_from_uri(MOCK_MEDIA_DIR_URI_1),
+            MOCK_MEDIA_DIR_URI_1,
+            id="legacy-data-scalar",
+        ),
+    ],
+)
+@pytest.mark.usefixtures("mock_media_source")
+async def test_reconfigure_flow(
+    hass: HomeAssistant,
+    entry_data: dict,
+    expected_uri: str,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test reconfigure flow loads the original data and can update media."""
+    freezer.move_to(TEST_TIME)
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        title="Test collection",
+        data=entry_data,
+    )
+    entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    state = hass.states.get("image.test_collection")
+    assert state and state.state == TEST_TIME
+
+    freezer.move_to(TEST_TIME_NEXT)
+    result = await entry.start_reconfigure_flow(hass)
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "reconfigure"
+    assert result["errors"] == {}
+
+    media_key = next(
+        key
+        for key in result["data_schema"].schema
+        if getattr(key, "schema", key) == CONF_MEDIA
+    )
+    assert (
+        media_key.description["suggested_value"][0]["media_content_id"] == expected_uri
+    )
+
+    # First try new data with error
+    new_data = data_from_uri([MOCK_MEDIA_DIR_URI_EMPTY])
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        new_data,
+    )
+
+    assert result.get("type") is FlowResultType.FORM
+    assert result.get("data") is None
+    assert result.get("errors") == {CONF_MEDIA: "selected_media_no_images"}
+
+    # Now update again with a valid option, to recover
+    new_data = data_from_uri([MOCK_MEDIA_DIR_URI_2])
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        new_data,
+    )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reconfigure_successful"
+
+    updated_entry = hass.config_entries.async_get_entry(entry.entry_id)
+    assert updated_entry is not None
+    assert updated_entry.data == new_data
+
+    await hass.async_block_till_done()
+
+    state = hass.states.get("image.test_collection")
+    assert state and state.state == TEST_TIME_NEXT