Abort if a flow is removed during a step (#142138)

* Abort if a flow is removed during a step

* Reorganize code

* Only call _set_pending_import_done if an entry is created

* Try a new approach

* Add tests

* Update tests

tests/test_data_entry_flow.py

@@ -243,6 +243,23 @@ async def test_abort_calls_async_remove(manager: MockFlowManager) -> None:
     assert len(manager.mock_created_entries) == 0
 
 
+async def test_abort_calls_async_flow_removed(manager: MockFlowManager) -> None:
+    """Test abort calling the async_flow_removed FlowManager method."""
+
+    @manager.mock_reg_handler("test")
+    class TestFlow(data_entry_flow.FlowHandler):
+        async def async_step_init(self, user_input=None):
+            return self.async_abort(reason="reason")
+
+    manager.async_flow_removed = Mock()
+    await manager.async_init("test")
+
+    manager.async_flow_removed.assert_called_once()
+
+    assert len(manager.async_progress()) == 0
+    assert len(manager.mock_created_entries) == 0
+
+
 async def test_abort_calls_async_remove_with_exception(
     manager: MockFlowManager, caplog: pytest.LogCaptureFixture
 ) -> None:
@@ -288,13 +305,7 @@ async def test_create_saves_data(manager: MockFlowManager) -> None:
 
 
 async def test_create_aborted_flow(manager: MockFlowManager) -> None:
-    """Test return create_entry from aborted flow.
-
-    Note: The entry is created even if the flow is already aborted, then the
-    flow raises an UnknownFlow exception. This behavior is not logical, and
-    we should consider changing it to not create the entry if the flow is
-    aborted.
-    """
+    """Test return create_entry from aborted flow."""
 
     @manager.mock_reg_handler("test")
     class TestFlow(data_entry_flow.FlowHandler):
@@ -308,14 +319,25 @@ async def test_create_aborted_flow(manager: MockFlowManager) -> None:
         await manager.async_init("test")
     assert len(manager.async_progress()) == 0
 
-    # The entry is created even if the flow is aborted
-    assert len(manager.mock_created_entries) == 1
+    # No entry should be created if the flow is aborted
+    assert len(manager.mock_created_entries) == 0
 
-    entry = manager.mock_created_entries[0]
-    assert entry["handler"] == "test"
-    assert entry["title"] == "Test Title"
-    assert entry["data"] == "Test Data"
-    assert entry["source"] is None
+
+async def test_create_calls_async_flow_removed(manager: MockFlowManager) -> None:
+    """Test create calling the async_flow_removed FlowManager method."""
+
+    @manager.mock_reg_handler("test")
+    class TestFlow(data_entry_flow.FlowHandler):
+        async def async_step_init(self, user_input=None):
+            return self.async_create_entry(title="Test Title", data="Test Data")
+
+    manager.async_flow_removed = Mock()
+    await manager.async_init("test")
+
+    manager.async_flow_removed.assert_called_once()
+
+    assert len(manager.async_progress()) == 0
+    assert len(manager.mock_created_entries) == 1
 
 
 async def test_discovery_init_flow(manager: MockFlowManager) -> None:
@@ -930,12 +952,34 @@ async def test_configure_raises_unknown_flow_if_not_in_progress(
         await manager.async_configure("wrong_flow_id")
 
 
-async def test_abort_raises_unknown_flow_if_not_in_progress(
+async def test_manager_abort_raises_unknown_flow_if_not_in_progress(
     manager: MockFlowManager,
 ) -> None:
     """Test abort raises UnknownFlow if the flow is not in progress."""
     with pytest.raises(data_entry_flow.UnknownFlow):
-        await manager.async_abort("wrong_flow_id")
+        manager.async_abort("wrong_flow_id")
+
+
+async def test_manager_abort_calls_async_flow_removed(manager: MockFlowManager) -> None:
+    """Test abort calling the async_flow_removed FlowManager method."""
+
+    @manager.mock_reg_handler("test")
+    class TestFlow(data_entry_flow.FlowHandler):
+        async def async_step_init(self, user_input=None):
+            return self.async_show_form(step_id="init")
+
+    manager.async_flow_removed = Mock()
+    result = await manager.async_init("test")
+    assert result["type"] == data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "init"
+
+    manager.async_flow_removed.assert_not_called()
+
+    manager.async_abort(result["flow_id"])
+    manager.async_flow_removed.assert_called_once()
+
+    assert len(manager.async_progress()) == 0
+    assert len(manager.mock_created_entries) == 0
 
 
 @pytest.mark.parametrize(