Avoid firing discovery events when flows immediately create a config entry (#155753)

homeassistant/config_entries.py

@@ -1473,16 +1473,18 @@ class ConfigEntriesFlowManager(
                 if not self._pending_import_flows[handler]:
                     del self._pending_import_flows[handler]
 
-        if (
-            result["type"] != data_entry_flow.FlowResultType.ABORT
-            and source in DISCOVERY_SOURCES
-        ):
+        # Flows can abort or create an entry in their initial step, we do not want to
+        # fire a discovery event if no flow is actually in progress
+        flow_completed = result["type"] in {
+            data_entry_flow.FlowResultType.ABORT,
+            data_entry_flow.FlowResultType.CREATE_ENTRY,
+        }
+
+        if not flow_completed and source in DISCOVERY_SOURCES:
             # Fire discovery event
             await self._discovery_event_debouncer.async_call()
 
-        if result["type"] != data_entry_flow.FlowResultType.ABORT and source in (
-            DISCOVERY_SOURCES | {SOURCE_REAUTH}
-        ):
+        if not flow_completed and source in DISCOVERY_SOURCES | {SOURCE_REAUTH}:
             # Notify listeners that a flow is created
             for subscription in self._flow_subscriptions:
                 subscription("added", flow.flow_id)

tests/components/config/test_config_entries.py

@@ -3,7 +3,7 @@
 from collections.abc import Generator
 from http import HTTPStatus
 from typing import Any
-from unittest.mock import ANY, AsyncMock, patch
+from unittest.mock import ANY, AsyncMock, Mock, patch
 
 from aiohttp.test_utils import TestClient
 from freezegun.api import FrozenDateTimeFactory
@@ -1002,6 +1002,36 @@ async def test_get_progress_subscribe(
         }
 
 
+async def test_get_progress_subscribe_create_entry(hass: HomeAssistant) -> None:
+    """Test flows creating entry immediately don't trigger subscription notification."""
+    assert await async_setup_component(hass, "config", {})
+    mock_platform(hass, "test.config_flow", None)
+
+    mock_integration(
+        hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True))
+    )
+
+    class TestFlow(core_ce.ConfigFlow):
+        VERSION = 1
+
+        async def async_step_import(
+            self, user_input: dict[str, Any]
+        ) -> ConfigFlowResult:
+            """Handle import - creates entry immediately."""
+            return self.async_create_entry(title="Test", data={})
+
+    subscription_mock = Mock()
+    hass.config_entries.flow.async_subscribe_flow(subscription_mock)
+
+    with mock_config_flow("test", TestFlow):
+        result = await hass.config_entries.flow.async_init(
+            "test", context={"source": core_ce.SOURCE_IMPORT}, data={}
+        )
+
+    assert result["type"] == FlowResultType.CREATE_ENTRY
+    assert len(subscription_mock.mock_calls) == 0
+
+
 async def test_get_progress_subscribe_in_progress(
     hass: HomeAssistant, hass_ws_client: WebSocketGenerator
 ) -> None: