Ensure config entry operations are always holding the lock (#117214)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
Co-authored-by: Paulus Schoutsen <balloob@gmail.com>

homeassistant/config_entries.py

@@ -523,8 +523,14 @@ class ConfigEntry(Generic[_DataT]):
             ):
                 raise OperationNotAllowed(
                     f"The config entry {self.title} ({self.domain}) with entry_id"
-                    f" {self.entry_id} cannot be setup because is already loaded in the"
+                    f" {self.entry_id} cannot be set up because it is already loaded "
-                    f" {self.state} state"
+                    f"in the {self.state} state"
+                )
+            if not self.setup_lock.locked():
+                raise OperationNotAllowed(
+                    f"The config entry {self.title} ({self.domain}) with entry_id"
+                    f" {self.entry_id} cannot be set up because it does not hold "
+                    "the setup lock"
                 )
             self._async_set_state(hass, ConfigEntryState.SETUP_IN_PROGRESS, None)
 
@@ -763,6 +769,13 @@ class ConfigEntry(Generic[_DataT]):
         component = await integration.async_get_component()
 
         if domain_is_integration := self.domain == integration.domain:
+            if not self.setup_lock.locked():
+                raise OperationNotAllowed(
+                    f"The config entry {self.title} ({self.domain}) with entry_id"
+                    f" {self.entry_id} cannot be unloaded because it does not hold "
+                    "the setup lock"
+                )
+
             if not self.state.recoverable:
                 return False
 
@@ -807,6 +820,13 @@ class ConfigEntry(Generic[_DataT]):
         if self.source == SOURCE_IGNORE:
             return
 
+        if not self.setup_lock.locked():
+            raise OperationNotAllowed(
+                f"The config entry {self.title} ({self.domain}) with entry_id"
+                f" {self.entry_id} cannot be removed because it does not hold "
+                "the setup lock"
+            )
+
         if not (integration := self._integration_for_domain):
             try:
                 integration = await loader.async_get_integration(hass, self.domain)
@@ -1639,7 +1659,7 @@ class ConfigEntries:
             if not entry.state.recoverable:
                 unload_success = entry.state is not ConfigEntryState.FAILED_UNLOAD
             else:
-                unload_success = await self.async_unload(entry_id)
+                unload_success = await self.async_unload(entry_id, _lock=False)
 
             await entry.async_remove(self.hass)
 
@@ -1741,7 +1761,7 @@ class ConfigEntries:
 
         self._entries = entries
 
-    async def async_setup(self, entry_id: str) -> bool:
+    async def async_setup(self, entry_id: str, _lock: bool = True) -> bool:
         """Set up a config entry.
 
         Return True if entry has been successfully loaded.
@@ -1752,13 +1772,17 @@ class ConfigEntries:
         if entry.state is not ConfigEntryState.NOT_LOADED:
             raise OperationNotAllowed(
                 f"The config entry {entry.title} ({entry.domain}) with entry_id"
-                f" {entry.entry_id} cannot be setup because is already loaded in the"
+                f" {entry.entry_id} cannot be set up because it is already loaded"
-                f" {entry.state} state"
+                f" in the {entry.state} state"
             )
 
         # Setup Component if not set up yet
         if entry.domain in self.hass.config.components:
-            await entry.async_setup(self.hass)
+            if _lock:
+                async with entry.setup_lock:
+                    await entry.async_setup(self.hass)
+            else:
+                await entry.async_setup(self.hass)
         else:
             # Setting up the component will set up all its config entries
             result = await async_setup_component(
@@ -1772,7 +1796,7 @@ class ConfigEntries:
             entry.state is ConfigEntryState.LOADED  # type: ignore[comparison-overlap]
         )
 
-    async def async_unload(self, entry_id: str) -> bool:
+    async def async_unload(self, entry_id: str, _lock: bool = True) -> bool:
         """Unload a config entry."""
         if (entry := self.async_get_entry(entry_id)) is None:
             raise UnknownEntry
@@ -1784,6 +1808,10 @@ class ConfigEntries:
                 f" recoverable state ({entry.state})"
             )
 
+        if _lock:
+            async with entry.setup_lock:
+                return await entry.async_unload(self.hass)
+
         return await entry.async_unload(self.hass)
 
     @callback
@@ -1825,12 +1853,12 @@ class ConfigEntries:
             return entry.state is ConfigEntryState.LOADED
 
         async with entry.setup_lock:
-            unload_result = await self.async_unload(entry_id)
+            unload_result = await self.async_unload(entry_id, _lock=False)
 
             if not unload_result or entry.disabled_by:
                 return unload_result
 
-            return await self.async_setup(entry_id)
+            return await self.async_setup(entry_id, _lock=False)
 
     async def async_set_disabled_by(
         self, entry_id: str, disabled_by: ConfigEntryDisabler | None

tests/components/aladdin_connect/test_init.py

@@ -138,7 +138,7 @@ async def test_load_and_unload(
     assert config_entry.state is ConfigEntryState.LOADED
     assert len(hass.config_entries.async_entries(DOMAIN)) == 1
 
-    assert await config_entry.async_unload(hass)
+    assert await hass.config_entries.async_unload(config_entry.entry_id)
     await hass.async_block_till_done()
     assert config_entry.state is ConfigEntryState.NOT_LOADED
 
@@ -218,7 +218,7 @@ async def test_stale_device_removal(
         for device in device_entries_other
     )
 
-    assert await config_entry.async_unload(hass)
+    assert await hass.config_entries.async_unload(config_entry.entry_id)
     await hass.async_block_till_done()
     assert config_entry.state is ConfigEntryState.NOT_LOADED
 

tests/components/azure_event_hub/conftest.py

@@ -63,7 +63,7 @@ async def mock_entry_fixture(hass, filter_schema, mock_create_batch, mock_send_b
 
     yield entry
 
-    await entry.async_unload(hass)
+    await hass.config_entries.async_unload(entry.entry_id)
 
 
 # fixtures for init tests

tests/components/deconz/test_init.py

@@ -150,7 +150,8 @@ async def test_unload_entry_multiple_gateways_parallel(
     assert len(hass.data[DECONZ_DOMAIN]) == 2
 
     await asyncio.gather(
-        config_entry.async_unload(hass), config_entry2.async_unload(hass)
+        hass.config_entries.async_unload(config_entry.entry_id),
+        hass.config_entries.async_unload(config_entry2.entry_id),
     )
 
     assert len(hass.data[DECONZ_DOMAIN]) == 0

tests/components/eafm/test_sensor.py

@@ -447,7 +447,7 @@ async def test_unload_entry(hass: HomeAssistant, mock_get_station) -> None:
     state = hass.states.get("sensor.my_station_water_level_stage")
     assert state.state == "5"
 
-    assert await entry.async_unload(hass)
+    await hass.config_entries.async_unload(entry.entry_id)
 
     # And the entity should be unavailable
     assert (

tests/components/energyzero/test_services.py

@@ -146,8 +146,7 @@ async def test_service_called_with_unloaded_entry(
     service: str,
 ) -> None:
     """Test service calls with unloaded config entry."""
-
+    await hass.config_entries.async_unload(mock_config_entry.entry_id)
-    await mock_config_entry.async_unload(hass)
 
     data = {"config_entry": mock_config_entry.entry_id, "incl_vat": True}
 

tests/components/fastdotcom/test_service.py

@@ -56,7 +56,7 @@ async def test_service_unloaded_entry(hass: HomeAssistant) -> None:
         await hass.async_block_till_done()
 
     assert config_entry
-    await config_entry.async_unload(hass)
+    await hass.config_entries.async_unload(config_entry.entry_id)
 
     with pytest.raises(HomeAssistantError) as exc:
         await hass.services.async_call(DOMAIN, SERVICE_NAME, blocking=True)

tests/components/forked_daapd/test_media_player.py

@@ -347,7 +347,7 @@ async def test_unload_config_entry(
     """Test the player is set unavailable when the config entry is unloaded."""
     assert hass.states.get(TEST_MASTER_ENTITY_NAME)
     assert hass.states.get(TEST_ZONE_ENTITY_NAMES[0])
-    await config_entry.async_unload(hass)
+    await hass.config_entries.async_unload(config_entry.entry_id)
     assert hass.states.get(TEST_MASTER_ENTITY_NAME).state == STATE_UNAVAILABLE
     assert hass.states.get(TEST_ZONE_ENTITY_NAMES[0]).state == STATE_UNAVAILABLE
 

tests/components/fully_kiosk/test_services.py

@@ -109,7 +109,7 @@ async def test_service_unloaded_entry(
     init_integration: MockConfigEntry,
 ) -> None:
     """Test service not called when config entry unloaded."""
-    await init_integration.async_unload(hass)
+    await hass.config_entries.async_unload(init_integration.entry_id)
 
     device_entry = device_registry.async_get_device(
         identifiers={(DOMAIN, "abcdef-123456")}

tests/components/heos/test_media_player.py

@@ -688,7 +688,7 @@ async def test_unload_config_entry(
 ) -> None:
     """Test the player is set unavailable when the config entry is unloaded."""
     await setup_platform(hass, config_entry, config)
-    await config_entry.async_unload(hass)
+    await hass.config_entries.async_unload(config_entry.entry_id)
     assert hass.states.get("media_player.test_player").state == STATE_UNAVAILABLE
 
 

tests/components/homekit_controller/test_init.py

@@ -70,7 +70,7 @@ async def test_async_remove_entry(hass: HomeAssistant) -> None:
     assert hkid in hass.data[ENTITY_MAP].storage_data
 
     # Remove it via config entry and number of pairings should go down
-    await helper.config_entry.async_remove(hass)
+    await hass.config_entries.async_remove(helper.config_entry.entry_id)
     assert len(controller.pairings) == 0
 
     assert hkid not in hass.data[ENTITY_MAP].storage_data

tests/components/homekit_controller/test_light.py

@@ -364,7 +364,7 @@ async def test_light_unloaded_removed(hass: HomeAssistant) -> None:
     state = await helper.poll_and_get_state()
     assert state.state == "off"
 
-    unload_result = await helper.config_entry.async_unload(hass)
+    unload_result = await hass.config_entries.async_unload(helper.config_entry.entry_id)
     assert unload_result is True
 
     # Make sure entity is set to unavailable state
@@ -374,11 +374,11 @@ async def test_light_unloaded_removed(hass: HomeAssistant) -> None:
     conn = hass.data[KNOWN_DEVICES]["00:00:00:00:00:00"]
     assert not conn.pollable_characteristics
 
-    await helper.config_entry.async_remove(hass)
+    await hass.config_entries.async_remove(helper.config_entry.entry_id)
     await hass.async_block_till_done()
 
     # Make sure entity is removed
-    assert hass.states.get(helper.entity_id).state == STATE_UNAVAILABLE
+    assert hass.states.get(helper.entity_id) is None
 
 
 async def test_migrate_unique_id(

tests/components/imap/test_init.py

@@ -76,7 +76,7 @@ async def test_entry_startup_and_unload(
     config_entry.add_to_hass(hass)
     assert await hass.config_entries.async_setup(config_entry.entry_id)
     await hass.async_block_till_done()
-    assert await config_entry.async_unload(hass)
+    assert await hass.config_entries.async_unload(config_entry.entry_id)
 
 
 @pytest.mark.parametrize(
@@ -449,7 +449,7 @@ async def test_handle_cleanup_exception(
     # Fail cleaning up
     mock_imap_protocol.close.side_effect = imap_close
 
-    assert await config_entry.async_unload(hass)
+    assert await hass.config_entries.async_unload(config_entry.entry_id)
     await hass.async_block_till_done()
     assert "Error while cleaning up imap connection" in caplog.text
 

tests/components/knx/test_services.py

@@ -290,7 +290,7 @@ async def test_reload_service(
 async def test_service_setup_failed(hass: HomeAssistant, knx: KNXTestKit) -> None:
     """Test service setup failed."""
     await knx.setup_integration({})
-    await knx.mock_config_entry.async_unload(hass)
+    await hass.config_entries.async_unload(knx.mock_config_entry.entry_id)
 
     with pytest.raises(HomeAssistantError) as exc_info:
         await hass.services.async_call(

tests/components/mqtt/test_common.py

@@ -1825,7 +1825,7 @@ async def help_test_reloadable(
     entry.add_to_hass(hass)
     mqtt_client_mock.connect.return_value = 0
     with patch("homeassistant.config.load_yaml_config_file", return_value=old_config):
-        await entry.async_setup(hass)
+        await hass.config_entries.async_setup(entry.entry_id)
 
     assert hass.states.get(f"{domain}.test_old_1")
     assert hass.states.get(f"{domain}.test_old_2")

tests/components/mqtt/test_init.py

@@ -1927,7 +1927,7 @@ async def test_reload_entry_with_restored_subscriptions(
     hass.config.components.add(mqtt.DOMAIN)
     mqtt_client_mock.connect.return_value = 0
     with patch("homeassistant.config.load_yaml_config_file", return_value={}):
-        await entry.async_setup(hass)
+        await hass.config_entries.async_setup(entry.entry_id)
 
     await mqtt.async_subscribe(hass, "test-topic", record_calls)
     await mqtt.async_subscribe(hass, "wild/+/card", record_calls)

tests/components/samsungtv/test_media_player.py

@@ -1369,7 +1369,7 @@ async def test_upnp_shutdown(
     state = hass.states.get(ENTITY_ID)
     assert state.state == STATE_ON
 
-    assert await entry.async_unload(hass)
+    assert await hass.config_entries.async_unload(entry.entry_id)
 
     state = hass.states.get(ENTITY_ID)
     assert state.state == STATE_UNAVAILABLE

tests/components/screenlogic/test_services.py

@@ -473,7 +473,7 @@ async def test_service_config_entry_not_loaded(
         assert hass.services.has_service(DOMAIN, SERVICE_SET_COLOR_MODE)
         assert len(hass.config_entries.async_entries(DOMAIN)) == 1
 
-        await mock_config_entry.async_unload(hass)
+        await hass.config_entries.async_unload(mock_config_entry.entry_id)
         await hass.async_block_till_done()
         assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
 

tests/components/vizio/test_init.py

@@ -43,7 +43,7 @@ async def test_tv_load_and_unload(
     assert len(hass.states.async_entity_ids(Platform.MEDIA_PLAYER)) == 1
     assert DOMAIN in hass.data
 
-    assert await config_entry.async_unload(hass)
+    assert await hass.config_entries.async_unload(config_entry.entry_id)
     await hass.async_block_till_done()
     entities = hass.states.async_entity_ids(Platform.MEDIA_PLAYER)
     assert len(entities) == 1
@@ -67,7 +67,7 @@ async def test_speaker_load_and_unload(
     assert len(hass.states.async_entity_ids(Platform.MEDIA_PLAYER)) == 1
     assert DOMAIN in hass.data
 
-    assert await config_entry.async_unload(hass)
+    assert await hass.config_entries.async_unload(config_entry.entry_id)
     await hass.async_block_till_done()
     entities = hass.states.async_entity_ids(Platform.MEDIA_PLAYER)
     assert len(entities) == 1

tests/components/ws66i/test_init.py

@@ -74,7 +74,7 @@ async def test_unload_config_entry(hass: HomeAssistant) -> None:
     assert hass.data[DOMAIN][config_entry.entry_id]
 
     with patch.object(MockWs66i, "close") as method_call:
-        await config_entry.async_unload(hass)
+        await hass.config_entries.async_unload(config_entry.entry_id)
         await hass.async_block_till_done()
 
         assert method_call.called

tests/components/yeelight/test_light.py

@@ -824,7 +824,7 @@ async def test_device_types(
         target_properties["music_mode"] = False
         assert dict(state.attributes) == target_properties
         await hass.config_entries.async_unload(config_entry.entry_id)
-        await config_entry.async_remove(hass)
+        await hass.config_entries.async_remove(config_entry.entry_id)
         registry = er.async_get(hass)
         registry.async_clear_config_entry(config_entry.entry_id)
         mocked_bulb.last_properties["nl_br"] = original_nightlight_brightness
@@ -846,7 +846,7 @@ async def test_device_types(
             assert dict(state.attributes) == nightlight_mode_properties
 
             await hass.config_entries.async_unload(config_entry.entry_id)
-            await config_entry.async_remove(hass)
+            await hass.config_entries.async_remove(config_entry.entry_id)
             registry.async_clear_config_entry(config_entry.entry_id)
             await hass.async_block_till_done()
             mocked_bulb.last_properties.pop("active_mode")
@@ -869,7 +869,7 @@ async def test_device_types(
             assert dict(state.attributes) == nightlight_entity_properties
 
             await hass.config_entries.async_unload(config_entry.entry_id)
-            await config_entry.async_remove(hass)
+            await hass.config_entries.async_remove(config_entry.entry_id)
             registry.async_clear_config_entry(config_entry.entry_id)
             await hass.async_block_till_done()
 

tests/components/zha/test_api.py

@@ -9,7 +9,6 @@ import pytest
 import zigpy.backups
 import zigpy.state
 
-from homeassistant.components import zha
 from homeassistant.components.zha import api
 from homeassistant.components.zha.core.const import RadioType
 from homeassistant.components.zha.core.helpers import get_zha_gateway
@@ -43,7 +42,7 @@ async def test_async_get_network_settings_inactive(
     await setup_zha()
 
     gateway = get_zha_gateway(hass)
-    await zha.async_unload_entry(hass, gateway.config_entry)
+    await hass.config_entries.async_unload(gateway.config_entry.entry_id)
 
     backup = zigpy.backups.NetworkBackup()
     backup.network_info.channel = 20
@@ -70,7 +69,7 @@ async def test_async_get_network_settings_missing(
     await setup_zha()
 
     gateway = get_zha_gateway(hass)
-    await gateway.config_entry.async_unload(hass)
+    await hass.config_entries.async_unload(gateway.config_entry.entry_id)
 
     # Network settings were never loaded for whatever reason
     zigpy_app_controller.state.network_info = zigpy.state.NetworkInfo()

tests/components/zha/test_discover.py

@@ -487,7 +487,7 @@ async def test_group_probe_cleanup_called(
     """Test cleanup happens when ZHA is unloaded."""
     await setup_zha()
     disc.GROUP_PROBE.cleanup = mock.Mock(wraps=disc.GROUP_PROBE.cleanup)
-    await config_entry.async_unload(hass_disable_services)
+    await hass_disable_services.config_entries.async_unload(config_entry.entry_id)
     await hass_disable_services.async_block_till_done()
     disc.GROUP_PROBE.cleanup.assert_called()
 

tests/conftest.py

@@ -566,7 +566,9 @@ async def hass(
         if loaded_entries:
             await asyncio.gather(
                 *(
-                    create_eager_task(config_entry.async_unload(hass))
+                    create_eager_task(
+                        hass.config_entries.async_unload(config_entry.entry_id)
+                    )
                     for config_entry in loaded_entries
                 )
             )

tests/test_config_entries.py

@@ -431,7 +431,7 @@ async def test_remove_entry_cancels_reauth(
     mock_platform(hass, "test.config_flow", None)
 
     entry.add_to_hass(hass)
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
 
     flows = hass.config_entries.flow.async_progress_by_handler("test")
@@ -472,7 +472,7 @@ async def test_remove_entry_handles_callback_error(
     # Check all config entries exist
     assert manager.async_entry_ids() == ["test1"]
     # Setup entry
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
 
     # Remove entry
@@ -1036,7 +1036,9 @@ async def test_reauth_notification(hass: HomeAssistant) -> None:
         assert "config_entry_reconfigure" not in notifications
 
 
-async def test_reauth_issue(hass: HomeAssistant) -> None:
+async def test_reauth_issue(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
     """Test that we create/delete an issue when source is reauth."""
     issue_registry = ir.async_get(hass)
     assert len(issue_registry.issues) == 0
@@ -1048,7 +1050,7 @@ async def test_reauth_issue(hass: HomeAssistant) -> None:
     mock_platform(hass, "test.config_flow", None)
 
     entry.add_to_hass(hass)
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
 
     flows = hass.config_entries.flow.async_progress_by_handler("test")
@@ -1175,10 +1177,13 @@ async def test_update_entry_options_and_trigger_listener(
 
 
 async def test_setup_raise_not_ready(
-    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    manager: config_entries.ConfigEntries,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test a setup raising not ready."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
 
     mock_setup_entry = AsyncMock(
         side_effect=ConfigEntryNotReady("The internet connection is offline")
@@ -1187,7 +1192,7 @@ async def test_setup_raise_not_ready(
     mock_platform(hass, "test.config_flow", None)
 
     with patch("homeassistant.config_entries.async_call_later") as mock_call:
-        await entry.async_setup(hass)
+        await manager.async_setup(entry.entry_id)
 
     assert len(mock_call.mock_calls) == 1
     assert (
@@ -1212,10 +1217,13 @@ async def test_setup_raise_not_ready(
 
 
 async def test_setup_raise_not_ready_from_exception(
-    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    manager: config_entries.ConfigEntries,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test a setup raising not ready from another exception."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
 
     original_exception = HomeAssistantError("The device dropped the connection")
     config_entry_exception = ConfigEntryNotReady()
@@ -1226,7 +1234,7 @@ async def test_setup_raise_not_ready_from_exception(
     mock_platform(hass, "test.config_flow", None)
 
     with patch("homeassistant.config_entries.async_call_later") as mock_call:
-        await entry.async_setup(hass)
+        await manager.async_setup(entry.entry_id)
 
     assert len(mock_call.mock_calls) == 1
     assert (
@@ -1235,29 +1243,35 @@ async def test_setup_raise_not_ready_from_exception(
     )
 
 
-async def test_setup_retrying_during_unload(hass: HomeAssistant) -> None:
+async def test_setup_retrying_during_unload(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
     """Test if we unload an entry that is in retry mode."""
     entry = MockConfigEntry(domain="test")
+    entry.add_to_hass(hass)
 
     mock_setup_entry = AsyncMock(side_effect=ConfigEntryNotReady)
     mock_integration(hass, MockModule("test", async_setup_entry=mock_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
     with patch("homeassistant.config_entries.async_call_later") as mock_call:
-        await entry.async_setup(hass)
+        await manager.async_setup(entry.entry_id)
 
     assert entry.state is config_entries.ConfigEntryState.SETUP_RETRY
     assert len(mock_call.return_value.mock_calls) == 0
 
-    await entry.async_unload(hass)
+    await manager.async_unload(entry.entry_id)
 
     assert entry.state is config_entries.ConfigEntryState.NOT_LOADED
     assert len(mock_call.return_value.mock_calls) == 1
 
 
-async def test_setup_retrying_during_unload_before_started(hass: HomeAssistant) -> None:
+async def test_setup_retrying_during_unload_before_started(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
     """Test if we unload an entry that is in retry mode before started."""
     entry = MockConfigEntry(domain="test")
+    entry.add_to_hass(hass)
     hass.set_state(CoreState.starting)
     initial_listeners = hass.bus.async_listeners()[EVENT_HOMEASSISTANT_STARTED]
 
@@ -1265,7 +1279,7 @@ async def test_setup_retrying_during_unload_before_started(hass: HomeAssistant)
     mock_integration(hass, MockModule("test", async_setup_entry=mock_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
 
     assert entry.state is config_entries.ConfigEntryState.SETUP_RETRY
@@ -1273,7 +1287,7 @@ async def test_setup_retrying_during_unload_before_started(hass: HomeAssistant)
         hass.bus.async_listeners()[EVENT_HOMEASSISTANT_STARTED] == initial_listeners + 1
     )
 
-    await entry.async_unload(hass)
+    await manager.async_unload(entry.entry_id)
     await hass.async_block_till_done()
 
     assert entry.state is config_entries.ConfigEntryState.NOT_LOADED
@@ -1282,15 +1296,18 @@ async def test_setup_retrying_during_unload_before_started(hass: HomeAssistant)
     )
 
 
-async def test_setup_does_not_retry_during_shutdown(hass: HomeAssistant) -> None:
+async def test_setup_does_not_retry_during_shutdown(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
     """Test we do not retry when HASS is shutting down."""
     entry = MockConfigEntry(domain="test")
+    entry.add_to_hass(hass)
 
     mock_setup_entry = AsyncMock(side_effect=ConfigEntryNotReady)
     mock_integration(hass, MockModule("test", async_setup_entry=mock_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
 
     assert entry.state is config_entries.ConfigEntryState.SETUP_RETRY
     assert len(mock_setup_entry.mock_calls) == 1
@@ -1693,6 +1710,98 @@ async def test_entry_cannot_be_loaded_twice(
     assert entry.state is state
 
 
+async def test_entry_setup_without_lock_raises(hass: HomeAssistant) -> None:
+    """Test trying to setup a config entry without the lock."""
+    entry = MockConfigEntry(
+        domain="comp", state=config_entries.ConfigEntryState.NOT_LOADED
+    )
+    entry.add_to_hass(hass)
+
+    async_setup = AsyncMock(return_value=True)
+    async_setup_entry = AsyncMock(return_value=True)
+    async_unload_entry = AsyncMock(return_value=True)
+
+    mock_integration(
+        hass,
+        MockModule(
+            "comp",
+            async_setup=async_setup,
+            async_setup_entry=async_setup_entry,
+            async_unload_entry=async_unload_entry,
+        ),
+    )
+    mock_platform(hass, "comp.config_flow", None)
+
+    with pytest.raises(
+        config_entries.OperationNotAllowed,
+        match="cannot be set up because it does not hold the setup lock",
+    ):
+        await entry.async_setup(hass)
+    assert len(async_setup.mock_calls) == 0
+    assert len(async_setup_entry.mock_calls) == 0
+    assert entry.state is config_entries.ConfigEntryState.NOT_LOADED
+
+
+async def test_entry_unload_without_lock_raises(hass: HomeAssistant) -> None:
+    """Test trying to unload a config entry without the lock."""
+    entry = MockConfigEntry(domain="comp", state=config_entries.ConfigEntryState.LOADED)
+    entry.add_to_hass(hass)
+
+    async_setup = AsyncMock(return_value=True)
+    async_setup_entry = AsyncMock(return_value=True)
+    async_unload_entry = AsyncMock(return_value=True)
+
+    mock_integration(
+        hass,
+        MockModule(
+            "comp",
+            async_setup=async_setup,
+            async_setup_entry=async_setup_entry,
+            async_unload_entry=async_unload_entry,
+        ),
+    )
+    mock_platform(hass, "comp.config_flow", None)
+
+    with pytest.raises(
+        config_entries.OperationNotAllowed,
+        match="cannot be unloaded because it does not hold the setup lock",
+    ):
+        await entry.async_unload(hass)
+    assert len(async_setup.mock_calls) == 0
+    assert len(async_setup_entry.mock_calls) == 0
+    assert entry.state is config_entries.ConfigEntryState.LOADED
+
+
+async def test_entry_remove_without_lock_raises(hass: HomeAssistant) -> None:
+    """Test trying to remove a config entry without the lock."""
+    entry = MockConfigEntry(domain="comp", state=config_entries.ConfigEntryState.LOADED)
+    entry.add_to_hass(hass)
+
+    async_setup = AsyncMock(return_value=True)
+    async_setup_entry = AsyncMock(return_value=True)
+    async_unload_entry = AsyncMock(return_value=True)
+
+    mock_integration(
+        hass,
+        MockModule(
+            "comp",
+            async_setup=async_setup,
+            async_setup_entry=async_setup_entry,
+            async_unload_entry=async_unload_entry,
+        ),
+    )
+    mock_platform(hass, "comp.config_flow", None)
+
+    with pytest.raises(
+        config_entries.OperationNotAllowed,
+        match="cannot be removed because it does not hold the setup lock",
+    ):
+        await entry.async_remove(hass)
+    assert len(async_setup.mock_calls) == 0
+    assert len(async_setup_entry.mock_calls) == 0
+    assert entry.state is config_entries.ConfigEntryState.LOADED
+
+
 @pytest.mark.parametrize(
     "state",
     [
@@ -3475,10 +3584,13 @@ async def test_entry_reload_calls_on_unload_listeners(
 
 
 async def test_setup_raise_entry_error(
-    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    manager: config_entries.ConfigEntries,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test a setup raising ConfigEntryError."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
 
     mock_setup_entry = AsyncMock(
         side_effect=ConfigEntryError("Incompatible firmware version")
@@ -3486,7 +3598,7 @@ async def test_setup_raise_entry_error(
     mock_integration(hass, MockModule("test", async_setup_entry=mock_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
     assert (
         "Error setting up entry test_title for test: Incompatible firmware version"
@@ -3498,10 +3610,13 @@ async def test_setup_raise_entry_error(
 
 
 async def test_setup_raise_entry_error_from_first_coordinator_update(
-    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    manager: config_entries.ConfigEntries,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test async_config_entry_first_refresh raises ConfigEntryError."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
 
     async def async_setup_entry(hass, entry):
         """Mock setup entry with a simple coordinator."""
@@ -3523,7 +3638,7 @@ async def test_setup_raise_entry_error_from_first_coordinator_update(
     mock_integration(hass, MockModule("test", async_setup_entry=async_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
     assert (
         "Error setting up entry test_title for test: Incompatible firmware version"
@@ -3535,10 +3650,13 @@ async def test_setup_raise_entry_error_from_first_coordinator_update(
 
 
 async def test_setup_not_raise_entry_error_from_future_coordinator_update(
-    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    manager: config_entries.ConfigEntries,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test a coordinator not raises ConfigEntryError in the future."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
 
     async def async_setup_entry(hass, entry):
         """Mock setup entry with a simple coordinator."""
@@ -3560,7 +3678,7 @@ async def test_setup_not_raise_entry_error_from_future_coordinator_update(
     mock_integration(hass, MockModule("test", async_setup_entry=async_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
     assert (
         "Config entry setup failed while fetching any data: Incompatible firmware"
@@ -3571,10 +3689,13 @@ async def test_setup_not_raise_entry_error_from_future_coordinator_update(
 
 
 async def test_setup_raise_auth_failed(
-    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    manager: config_entries.ConfigEntries,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test a setup raising ConfigEntryAuthFailed."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
 
     mock_setup_entry = AsyncMock(
         side_effect=ConfigEntryAuthFailed("The password is no longer valid")
@@ -3582,7 +3703,7 @@ async def test_setup_raise_auth_failed(
     mock_integration(hass, MockModule("test", async_setup_entry=mock_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
     assert "could not authenticate: The password is no longer valid" in caplog.text
 
@@ -3597,7 +3718,7 @@ async def test_setup_raise_auth_failed(
     caplog.clear()
     entry._async_set_state(hass, config_entries.ConfigEntryState.NOT_LOADED, None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
     assert "could not authenticate: The password is no longer valid" in caplog.text
 
@@ -3608,10 +3729,13 @@ async def test_setup_raise_auth_failed(
 
 
 async def test_setup_raise_auth_failed_from_first_coordinator_update(
-    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    manager: config_entries.ConfigEntries,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test async_config_entry_first_refresh raises ConfigEntryAuthFailed."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
 
     async def async_setup_entry(hass, entry):
         """Mock setup entry with a simple coordinator."""
@@ -3633,7 +3757,7 @@ async def test_setup_raise_auth_failed_from_first_coordinator_update(
     mock_integration(hass, MockModule("test", async_setup_entry=async_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
     assert "could not authenticate: The password is no longer valid" in caplog.text
 
@@ -3646,7 +3770,7 @@ async def test_setup_raise_auth_failed_from_first_coordinator_update(
     caplog.clear()
     entry._async_set_state(hass, config_entries.ConfigEntryState.NOT_LOADED, None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
     assert "could not authenticate: The password is no longer valid" in caplog.text
 
@@ -3657,10 +3781,13 @@ async def test_setup_raise_auth_failed_from_first_coordinator_update(
 
 
 async def test_setup_raise_auth_failed_from_future_coordinator_update(
-    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+    hass: HomeAssistant,
+    manager: config_entries.ConfigEntries,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test a coordinator raises ConfigEntryAuthFailed in the future."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
 
     async def async_setup_entry(hass, entry):
         """Mock setup entry with a simple coordinator."""
@@ -3682,7 +3809,7 @@ async def test_setup_raise_auth_failed_from_future_coordinator_update(
     mock_integration(hass, MockModule("test", async_setup_entry=async_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
     assert "Authentication failed while fetching" in caplog.text
     assert "The password is no longer valid" in caplog.text
@@ -3696,7 +3823,7 @@ async def test_setup_raise_auth_failed_from_future_coordinator_update(
     caplog.clear()
     entry._async_set_state(hass, config_entries.ConfigEntryState.NOT_LOADED, None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
     assert "Authentication failed while fetching" in caplog.text
     assert "The password is no longer valid" in caplog.text
@@ -3719,16 +3846,19 @@ async def test_initialize_and_shutdown(hass: HomeAssistant) -> None:
     assert mock_async_shutdown.called
 
 
-async def test_setup_retrying_during_shutdown(hass: HomeAssistant) -> None:
+async def test_setup_retrying_during_shutdown(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
     """Test if we shutdown an entry that is in retry mode."""
     entry = MockConfigEntry(domain="test")
+    entry.add_to_hass(hass)
 
     mock_setup_entry = AsyncMock(side_effect=ConfigEntryNotReady)
     mock_integration(hass, MockModule("test", async_setup_entry=mock_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
     with patch("homeassistant.helpers.event.async_call_later") as mock_call:
-        await entry.async_setup(hass)
+        await manager.async_setup(entry.entry_id)
 
     assert entry.state is config_entries.ConfigEntryState.SETUP_RETRY
     assert len(mock_call.return_value.mock_calls) == 0
@@ -3747,7 +3877,9 @@ async def test_setup_retrying_during_shutdown(hass: HomeAssistant) -> None:
     entry.async_cancel_retry_setup()
 
 
-async def test_scheduling_reload_cancels_setup_retry(hass: HomeAssistant) -> None:
+async def test_scheduling_reload_cancels_setup_retry(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
     """Test scheduling a reload cancels setup retry."""
     entry = MockConfigEntry(domain="test")
     entry.add_to_hass(hass)
@@ -3760,7 +3892,7 @@ async def test_scheduling_reload_cancels_setup_retry(hass: HomeAssistant) -> Non
     with patch(
         "homeassistant.config_entries.async_call_later", return_value=cancel_mock
     ):
-        await entry.async_setup(hass)
+        await manager.async_setup(entry.entry_id)
 
     assert entry.state is config_entries.ConfigEntryState.SETUP_RETRY
     assert len(cancel_mock.mock_calls) == 0
@@ -4190,16 +4322,20 @@ async def test_disallow_entry_reload_with_setup_in_progress(
     assert entry.state is config_entries.ConfigEntryState.SETUP_IN_PROGRESS
 
 
-async def test_reauth(hass: HomeAssistant) -> None:
+async def test_reauth(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
     """Test the async_reauth_helper."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
     entry2 = MockConfigEntry(title="test_title", domain="test")
+    entry2.add_to_hass(hass)
 
     mock_setup_entry = AsyncMock(return_value=True)
     mock_integration(hass, MockModule("test", async_setup_entry=mock_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
 
     flow = hass.config_entries.flow
@@ -4252,16 +4388,20 @@ async def test_reauth(hass: HomeAssistant) -> None:
     assert len(hass.config_entries.flow.async_progress()) == 1
 
 
-async def test_reconfigure(hass: HomeAssistant) -> None:
+async def test_reconfigure(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
     """Test the async_reconfigure_helper."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
     entry2 = MockConfigEntry(title="test_title", domain="test")
+    entry2.add_to_hass(hass)
 
     mock_setup_entry = AsyncMock(return_value=True)
     mock_integration(hass, MockModule("test", async_setup_entry=mock_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
 
     flow = hass.config_entries.flow
@@ -4340,14 +4480,17 @@ async def test_reconfigure(hass: HomeAssistant) -> None:
     assert len(hass.config_entries.flow.async_progress()) == 1
 
 
-async def test_get_active_flows(hass: HomeAssistant) -> None:
+async def test_get_active_flows(
+    hass: HomeAssistant, manager: config_entries.ConfigEntries
+) -> None:
     """Test the async_get_active_flows helper."""
     entry = MockConfigEntry(title="test_title", domain="test")
+    entry.add_to_hass(hass)
     mock_setup_entry = AsyncMock(return_value=True)
     mock_integration(hass, MockModule("test", async_setup_entry=mock_setup_entry))
     mock_platform(hass, "test.config_flow", None)
 
-    await entry.async_setup(hass)
+    await manager.async_setup(entry.entry_id)
     await hass.async_block_till_done()
 
     flow = hass.config_entries.flow