Add `single_config_entry` manifest option (#109505)

* Allow setting if we support multiple config entries in config flow

* Move property to config flow instead of flow handler

* Move marking an integration as single instance only to manifest

* Revert line remove

* Avoid init a config flow or adding a new entry on a single instance with an entry

* Revert changes in test

* Process code review comments

* Apply code review suggestion

homeassistant/config_entries.py

@@ -1007,6 +1007,19 @@ class ConfigEntriesFlowManager(data_entry_flow.FlowManager):
         if not context or "source" not in context:
             raise KeyError("Context not set or doesn't have a source set")
 
+        # Avoid starting a config flow on an integration that only supports
+        # a single config entry, but which already has an entry
+        if (
+            context.get("source") != SOURCE_REAUTH
+            and await _support_single_config_entry_only(self.hass, handler)
+            and self.config_entries.async_has_entry(handler)
+        ):
+            raise HomeAssistantError(
+                "Cannot start a config flow, the integration"
+                " supports only a single config entry"
+                " but already has one"
+            )
+
         flow_id = uuid_util.random_uuid_hex()
         loop = self.hass.loop
 
@@ -1104,11 +1117,21 @@ class ConfigEntriesFlowManager(data_entry_flow.FlowManager):
         # or the default discovery ID
         for progress_flow in self.async_progress_by_handler(flow.handler):
             progress_unique_id = progress_flow["context"].get("unique_id")
-            if progress_flow["flow_id"] != flow.flow_id and (
+            progress_flow_id = progress_flow["flow_id"]
+
+            if progress_flow_id != flow.flow_id and (
                 (flow.unique_id and progress_unique_id == flow.unique_id)
                 or progress_unique_id == DEFAULT_DISCOVERY_UNIQUE_ID
             ):
-                self.async_abort(progress_flow["flow_id"])
+                self.async_abort(progress_flow_id)
+
+            # Abort any flows in progress for the same handler
+            # when integration allows only one config entry
+            if (
+                progress_flow_id != flow.flow_id
+                and await _support_single_config_entry_only(self.hass, flow.handler)
+            ):
+                self.async_abort(progress_flow_id)
 
         if flow.unique_id is not None:
             # Reset unique ID when the default discovery ID has been used
@@ -1367,12 +1390,28 @@ class ConfigEntries:
         """Return entry for a domain with a matching unique id."""
         return self._entries.get_entry_by_domain_and_unique_id(domain, unique_id)
 
+    @callback
+    def async_has_entry(self, domain: str) -> bool:
+        """Return if there are entries for a domain."""
+        return bool(self.async_entries(domain))
+
     async def async_add(self, entry: ConfigEntry) -> None:
         """Add and setup an entry."""
         if entry.entry_id in self._entries.data:
             raise HomeAssistantError(
                 f"An entry with the id {entry.entry_id} already exists."
             )
+
+        # Avoid adding a config entry for a integration
+        # that only supports a single config entry, but already has an entry
+        if await _support_single_config_entry_only(
+            self.hass, entry.domain
+        ) and self.async_has_entry(entry.domain):
+            raise HomeAssistantError(
+                f"An entry for {entry.domain} already exists,"
+                f" but integration supports only one config entry"
+            )
+
         self._entries[entry.entry_id] = entry
         self._async_dispatch(ConfigEntryChange.ADDED, entry)
         await self.async_setup(entry.entry_id)
@@ -2371,6 +2410,12 @@ async def support_remove_from_device(hass: HomeAssistant, domain: str) -> bool:
     return hasattr(component, "async_remove_config_entry_device")
 
 
+async def _support_single_config_entry_only(hass: HomeAssistant, domain: str) -> bool:
+    """Test if a domain supports only a single config entry."""
+    integration = await loader.async_get_integration(hass, domain)
+    return integration.single_config_entry
+
+
 async def _load_integration(
     hass: HomeAssistant, domain: str, hass_config: ConfigType
 ) -> None:

homeassistant/loader.py

@@ -179,6 +179,7 @@ class Manifest(TypedDict, total=False):
     version: str
     codeowners: list[str]
     loggers: list[str]
+    single_config_entry: bool
 
 
 def async_setup(hass: HomeAssistant) -> None:
@@ -368,6 +369,9 @@ async def async_get_integration_descriptions(
             "integration_type": integration.integration_type,
             "iot_class": integration.iot_class,
             "name": integration.name,
+            "single_config_entry": integration.manifest.get(
+                "single_config_entry", False
+            ),
         }
         custom_flows[integration_key][integration.domain] = metadata
 
@@ -770,6 +774,11 @@ class Integration:
             return None
         return AwesomeVersion(self.manifest["version"])
 
+    @cached_property
+    def single_config_entry(self) -> bool:
+        """Return if the integration supports a single config entry only."""
+        return self.manifest.get("single_config_entry", False)
+
     @property
     def all_dependencies(self) -> set[str]:
         """Return all dependencies including sub-dependencies."""

script/hassfest/config_flow.py

@@ -191,6 +191,11 @@ def _generate_integrations(
                 if integration.iot_class:
                     metadata["iot_class"] = integration.iot_class
 
+                if single_config_entry := integration.manifest.get(
+                    "single_config_entry"
+                ):
+                    metadata["single_config_entry"] = single_config_entry
+
             if integration.integration_type == "helper":
                 result["helper"][domain] = metadata
             else:

script/hassfest/manifest.py

@@ -265,6 +265,7 @@ INTEGRATION_MANIFEST_SCHEMA = vol.Schema(
         vol.Optional("loggers"): [str],
         vol.Optional("disabled"): str,
         vol.Optional("iot_class"): vol.In(SUPPORTED_IOT_CLASSES),
+        vol.Optional("single_config_entry"): bool,
     }
 )
 

tests/test_config_entries.py

@@ -4434,6 +4434,177 @@ async def test_hashable_non_string_unique_id(
     assert entries.get_entry_by_domain_and_unique_id("test", unique_id) is None
 
 
+async def test_avoid_starting_config_flow_on_single_config_entry(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
+    """Test that we cannot start a config flow for a single config entry integration."""
+    integration = loader.Integration(
+        hass,
+        "components.comp",
+        None,
+        {
+            "name": "Comp",
+            "dependencies": [],
+            "requirements": [],
+            "domain": "comp",
+            "single_config_entry": True,
+        },
+    )
+    entry = MockConfigEntry(
+        domain="comp",
+        unique_id="1234",
+        title="Test",
+        data={"vendor": "data"},
+        options={"vendor": "options"},
+    )
+    entry.add_to_hass(hass)
+
+    mock_platform(hass, "comp.config_flow", None)
+
+    with patch(
+        "homeassistant.loader.async_get_integration",
+        return_value=integration,
+    ), pytest.raises(
+        HomeAssistantError,
+        match=r"Cannot start a config flow, the integration supports only a single config entry but already has one",
+    ):
+        await hass.config_entries.flow.async_init("comp", context={"source": "user"})
+
+
+async def test_avoid_adding_second_config_entry_on_single_config_entry(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
+    """Test that we cannot add a second entry for a single config entry integration."""
+    integration = loader.Integration(
+        hass,
+        "components.comp",
+        None,
+        {
+            "name": "Comp",
+            "dependencies": [],
+            "requirements": [],
+            "domain": "comp",
+            "single_config_entry": True,
+        },
+    )
+    entry = MockConfigEntry(
+        domain="comp",
+        unique_id="1234",
+        title="Test",
+        data={"vendor": "data"},
+        options={"vendor": "options"},
+    )
+    entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.loader.async_get_integration",
+        return_value=integration,
+    ), pytest.raises(
+        HomeAssistantError,
+        match=r"An entry for comp already exists, but integration supports only one config entry",
+    ):
+        await hass.config_entries.async_add(
+            MockConfigEntry(
+                title="Second comp entry", domain="comp", data={"vendor": "data2"}
+            )
+        )
+
+
+async def test_in_progress_get_canceled_when_entry_is_created(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
+    """Test that we abort all in progress flows when a new entry is created on a single instance only integration."""
+    integration = loader.Integration(
+        hass,
+        "components.comp",
+        None,
+        {
+            "name": "Comp",
+            "dependencies": [],
+            "requirements": [],
+            "domain": "comp",
+            "single_config_entry": True,
+        },
+    )
+    mock_integration(hass, MockModule("comp"))
+    mock_platform(hass, "comp.config_flow", None)
+
+    class TestFlow(config_entries.ConfigFlow):
+        """Test flow."""
+
+        VERSION = 1
+
+        async def async_step_user(self, user_input=None):
+            """Test user step."""
+            if user_input is not None:
+                return self.async_create_entry(title="Test Title", data=user_input)
+
+            return self.async_show_form(step_id="user")
+
+    with patch.dict(config_entries.HANDLERS, {"comp": TestFlow}), patch(
+        "homeassistant.loader.async_get_integration",
+        return_value=integration,
+    ):
+        # Create one to be in progress
+        result = await manager.flow.async_init(
+            "comp", context={"source": config_entries.SOURCE_USER}
+        )
+        assert result["type"] == data_entry_flow.FlowResultType.FORM
+
+        # Will be canceled
+        result2 = await manager.flow.async_init(
+            "comp", context={"source": config_entries.SOURCE_USER}
+        )
+        assert result2["type"] == data_entry_flow.FlowResultType.FORM
+
+        result = await manager.flow.async_configure(
+            result["flow_id"], user_input={"host": "127.0.0.1"}
+        )
+        assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
+
+    assert len(manager.flow.async_progress()) == 0
+    assert len(manager.async_entries()) == 1
+
+
+async def test_start_reauth_still_possible_for_single_config_entry(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+):
+    """Test that we can still start a reauth flow on a single config entry integration."""
+    integration = loader.Integration(
+        hass,
+        "components.comp",
+        None,
+        {
+            "name": "Comp",
+            "dependencies": [],
+            "requirements": [],
+            "domain": "comp",
+            "single_config_entry": True,
+        },
+    )
+    entry = MockConfigEntry(
+        domain="comp",
+        unique_id="1234",
+        title="Test",
+        data={"vendor": "data"},
+        options={"vendor": "options"},
+    )
+    entry.add_to_hass(hass)
+
+    mock_platform(hass, "comp.config_flow", None)
+
+    with patch(
+        "homeassistant.loader.async_get_integration",
+        return_value=integration,
+    ):
+        result = await hass.config_entries.flow.async_init(
+            "comp", context={"source": config_entries.SOURCE_REAUTH}
+        )
+
+    assert result["type"] == data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "reauth_confirm"
+
+
 async def test_directly_mutating_blocked(
     hass: HomeAssistant, caplog: pytest.LogCaptureFixture
 ) -> None: