Prompt user to remove application credentials when deleting config entries (#74825)

* Prompt user to remove application credentials when deleting config entries

* Adjust assertions on intermediate state in config entry tests

* Add a callback hook to modify config entry remove result

* Improve test coverage and simplify implementation

* Register remove callback per domain

* Update homeassistant/components/application_credentials/__init__.py

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

* Fix tests to use new variable name including domain

* Add websocket command to return application credentials for an integration

* Remove unnecessary diff

* Apply suggestions from code review

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

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

homeassistant/components/application_credentials/__init__.py

@@ -15,6 +15,7 @@ import voluptuous as vol
 
 from homeassistant.components import websocket_api
 from homeassistant.components.websocket_api.connection import ActiveConnection
+from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
     CONF_CLIENT_ID,
     CONF_CLIENT_SECRET,
@@ -127,9 +128,7 @@ class ApplicationCredentialsStorageCollection(collection.StorageCollection):
         for item in self.async_items():
             if item[CONF_DOMAIN] != domain:
                 continue
-            auth_domain = (
-                item[CONF_AUTH_DOMAIN] if CONF_AUTH_DOMAIN in item else item[CONF_ID]
-            )
+            auth_domain = item.get(CONF_AUTH_DOMAIN, item[CONF_ID])
             credentials[auth_domain] = ClientCredential(
                 client_id=item[CONF_CLIENT_ID],
                 client_secret=item[CONF_CLIENT_SECRET],
@@ -156,6 +155,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     ).async_setup(hass)
 
     websocket_api.async_register_command(hass, handle_integration_list)
+    websocket_api.async_register_command(hass, handle_config_entry)
 
     config_entry_oauth2_flow.async_add_implementation_provider(
         hass, DOMAIN, _async_provide_implementation
@@ -234,6 +234,27 @@ async def _async_provide_implementation(
     ]
 
 
+async def _async_config_entry_app_credentials(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+) -> str | None:
+    """Return the item id of an application credential for an existing ConfigEntry."""
+    if not await _get_platform(hass, config_entry.domain) or not (
+        auth_domain := config_entry.data.get("auth_implementation")
+    ):
+        return None
+
+    storage_collection = hass.data[DOMAIN][DATA_STORAGE]
+    for item in storage_collection.async_items():
+        item_id = item[CONF_ID]
+        if (
+            item[CONF_DOMAIN] == config_entry.domain
+            and item.get(CONF_AUTH_DOMAIN, item_id) == auth_domain
+        ):
+            return item_id
+    return None
+
+
 class ApplicationCredentialsProtocol(Protocol):
     """Define the format that application_credentials platforms may have.
 
@@ -311,3 +332,31 @@ async def handle_integration_list(
         },
     }
     connection.send_result(msg["id"], result)
+
+
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "application_credentials/config_entry",
+        vol.Required("config_entry_id"): str,
+    }
+)
+@websocket_api.async_response
+async def handle_config_entry(
+    hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
+) -> None:
+    """Return application credentials information for a config entry."""
+    entry_id = msg["config_entry_id"]
+    config_entry = hass.config_entries.async_get_entry(entry_id)
+    if not config_entry:
+        connection.send_error(
+            msg["id"],
+            "invalid_config_entry_id",
+            f"Config entry not found: {entry_id}",
+        )
+        return
+    result = {}
+    if application_credentials_id := await _async_config_entry_app_credentials(
+        hass, config_entry
+    ):
+        result["application_credentials_id"] = application_credentials_id
+    connection.send_result(msg["id"], result)

tests/components/application_credentials/test_init.py

@@ -30,7 +30,7 @@ from homeassistant.core import HomeAssistant
 from homeassistant.helpers import config_entry_oauth2_flow
 from homeassistant.setup import async_setup_component
 
-from tests.common import mock_platform
+from tests.common import MockConfigEntry, mock_platform
 
 CLIENT_ID = "some-client-id"
 CLIENT_SECRET = "some-client-secret"
@@ -90,10 +90,12 @@ async def mock_application_credentials_integration(
     authorization_server: AuthorizationServer,
 ):
     """Mock a application_credentials integration."""
-    assert await async_setup_component(hass, "application_credentials", {})
-    await setup_application_credentials_integration(
-        hass, TEST_DOMAIN, authorization_server
-    )
+    with patch("homeassistant.loader.APPLICATION_CREDENTIALS", [TEST_DOMAIN]):
+        assert await async_setup_component(hass, "application_credentials", {})
+        await setup_application_credentials_integration(
+            hass, TEST_DOMAIN, authorization_server
+        )
+        yield
 
 
 class FakeConfigFlow(config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN):
@@ -418,7 +420,7 @@ async def test_config_flow_no_credentials(hass):
         TEST_DOMAIN, context={"source": config_entries.SOURCE_USER}
     )
     assert result.get("type") == data_entry_flow.FlowResultType.ABORT
-    assert result.get("reason") == "missing_configuration"
+    assert result.get("reason") == "missing_credentials"
 
 
 async def test_config_flow_other_domain(
@@ -445,7 +447,7 @@ async def test_config_flow_other_domain(
         TEST_DOMAIN, context={"source": config_entries.SOURCE_USER}
     )
     assert result.get("type") == data_entry_flow.FlowResultType.ABORT
-    assert result.get("reason") == "missing_configuration"
+    assert result.get("reason") == "missing_credentials"
 
 
 async def test_config_flow(
@@ -483,6 +485,27 @@ async def test_config_flow(
         == "Cannot delete credential in use by integration fake_integration"
     )
 
+    # Return information about the in use config entry
+    entries = hass.config_entries.async_entries(TEST_DOMAIN)
+    assert len(entries) == 1
+    client = await ws_client()
+    result = await client.cmd_result(
+        "config_entry", {"config_entry_id": entries[0].entry_id}
+    )
+    assert result.get("application_credentials_id") == ID
+
+    # Delete the config entry
+    await hass.config_entries.async_remove(entries[0].entry_id)
+
+    # Application credential can now be removed
+    resp = await client.cmd("delete", {"application_credentials_id": ID})
+    assert resp.get("success")
+
+    # Config entry information no longer found
+    result = await client.cmd("config_entry", {"config_entry_id": entries[0].entry_id})
+    assert "error" in result
+    assert result["error"].get("code") == "invalid_config_entry_id"
+
 
 async def test_config_flow_multiple_entries(
     hass: HomeAssistant,
@@ -549,7 +572,7 @@ async def test_config_flow_create_delete_credential(
         TEST_DOMAIN, context={"source": config_entries.SOURCE_USER}
     )
     assert result.get("type") == data_entry_flow.FlowResultType.ABORT
-    assert result.get("reason") == "missing_configuration"
+    assert result.get("reason") == "missing_credentials"
 
 
 @pytest.mark.parametrize("config_credential", [DEVELOPER_CREDENTIAL])
@@ -751,3 +774,24 @@ async def test_name(
     assert (
         result["data"].get("auth_implementation") == "fake_integration_some_client_id"
     )
+
+
+async def test_remove_config_entry_without_app_credentials(
+    hass: HomeAssistant,
+    ws_client: ClientFixture,
+    authorization_server: AuthorizationServer,
+):
+    """Test config entry removal for non-app credentials integration."""
+    hass.config.components.add("other_domain")
+    config_entry = MockConfigEntry(domain="other_domain")
+    config_entry.add_to_hass(hass)
+    assert await async_setup_component(hass, "other_domain", {})
+
+    entries = hass.config_entries.async_entries("other_domain")
+    assert len(entries) == 1
+
+    client = await ws_client()
+    result = await client.cmd_result(
+        "config_entry", {"config_entry_id": entries[0].entry_id}
+    )
+    assert "application_credential_id" not in result