Google Assistant SDK integration (#82328)

* Copy google_sheets to google_assistant_sdk

This is to improve diff of the next commit with the actual implementation.

Commands used:
cp -r homeassistant/components/google_sheets/ homeassistant/components/google_assistant_sdk/
cp -r tests/components/google_sheets/ tests/components/google_assistant_sdk/

find homeassistant/components/google_assistant_sdk/ tests/components/google_assistant_sdk/ -type f | xargs sed -i \
-e 's@google_sheets@google_assistant_sdk@g' \
-e 's@Google Sheets@Google Assistant SDK@g' \
-e 's@tkdrob@tronikos@g'

* Google Assistant SDK integration
Allows sending commands and broadcast messages to Google Assistant.

* Remove unnecessary async_entry_has_scopes check

* Bump gassist-text to fix protobuf dependency

CODEOWNERS

@@ -428,6 +428,8 @@ build.json @home-assistant/supervisor
 /tests/components/google/ @allenporter
 /homeassistant/components/google_assistant/ @home-assistant/cloud
 /tests/components/google_assistant/ @home-assistant/cloud
+/homeassistant/components/google_assistant_sdk/ @tronikos
+/tests/components/google_assistant_sdk/ @tronikos
 /homeassistant/components/google_cloud/ @lufton
 /homeassistant/components/google_sheets/ @tkdrob
 /tests/components/google_sheets/ @tkdrob

homeassistant/brands/google.json

@@ -3,6 +3,7 @@
   "name": "Google",
   "integrations": [
     "google_assistant",
+    "google_assistant_sdk",
     "google_cloud",
     "google_domains",
     "google_maps",

homeassistant/components/google_assistant_sdk/__init__.py

@@ -0,0 +1,92 @@
+"""Support for Google Assistant SDK."""
+from __future__ import annotations
+
+import aiohttp
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigEntry, ConfigEntryState
+from homeassistant.const import CONF_NAME, Platform
+from homeassistant.core import HomeAssistant, ServiceCall
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+from homeassistant.helpers import discovery
+from homeassistant.helpers.config_entry_oauth2_flow import (
+    OAuth2Session,
+    async_get_config_entry_implementation,
+)
+from homeassistant.helpers.typing import ConfigType
+
+from .const import DOMAIN
+from .helpers import async_send_text_commands
+
+SERVICE_SEND_TEXT_COMMAND = "send_text_command"
+SERVICE_SEND_TEXT_COMMAND_FIELD_COMMAND = "command"
+SERVICE_SEND_TEXT_COMMAND_SCHEMA = vol.All(
+    {
+        vol.Required(SERVICE_SEND_TEXT_COMMAND_FIELD_COMMAND): vol.All(
+            str, vol.Length(min=1)
+        ),
+    },
+)
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up Google Assistant SDK component."""
+    hass.async_create_task(
+        discovery.async_load_platform(
+            hass, Platform.NOTIFY, DOMAIN, {CONF_NAME: DOMAIN}, config
+        )
+    )
+
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Google Assistant SDK from a config entry."""
+    implementation = await async_get_config_entry_implementation(hass, entry)
+    session = OAuth2Session(hass, entry, implementation)
+    try:
+        await session.async_ensure_token_valid()
+    except aiohttp.ClientResponseError as err:
+        if 400 <= err.status < 500:
+            raise ConfigEntryAuthFailed(
+                "OAuth session is not valid, reauth required"
+            ) from err
+        raise ConfigEntryNotReady from err
+    except aiohttp.ClientError as err:
+        raise ConfigEntryNotReady from err
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = session
+
+    await async_setup_service(hass)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    hass.data[DOMAIN].pop(entry.entry_id)
+    loaded_entries = [
+        entry
+        for entry in hass.config_entries.async_entries(DOMAIN)
+        if entry.state == ConfigEntryState.LOADED
+    ]
+    if len(loaded_entries) == 1:
+        for service_name in hass.services.async_services()[DOMAIN]:
+            hass.services.async_remove(DOMAIN, service_name)
+
+    return True
+
+
+async def async_setup_service(hass: HomeAssistant) -> None:
+    """Add the services for Google Assistant SDK."""
+
+    async def send_text_command(call: ServiceCall) -> None:
+        """Send a text command to Google Assistant SDK."""
+        command: str = call.data[SERVICE_SEND_TEXT_COMMAND_FIELD_COMMAND]
+        await async_send_text_commands([command], hass)
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_SEND_TEXT_COMMAND,
+        send_text_command,
+        schema=SERVICE_SEND_TEXT_COMMAND_SCHEMA,
+    )

homeassistant/components/google_assistant_sdk/application_credentials.py

@@ -0,0 +1,22 @@
+"""application_credentials platform for Google Assistant SDK."""
+import oauth2client
+
+from homeassistant.components.application_credentials import AuthorizationServer
+from homeassistant.core import HomeAssistant
+
+
+async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
+    """Return authorization server."""
+    return AuthorizationServer(
+        oauth2client.GOOGLE_AUTH_URI,
+        oauth2client.GOOGLE_TOKEN_URI,
+    )
+
+
+async def async_get_description_placeholders(hass: HomeAssistant) -> dict[str, str]:
+    """Return description placeholders for the credentials dialog."""
+    return {
+        "oauth_consent_url": "https://console.cloud.google.com/apis/credentials/consent",
+        "more_info_url": "https://www.home-assistant.io/integrations/google_assistant_sdk/",
+        "oauth_creds_url": "https://console.cloud.google.com/apis/credentials",
+    }

homeassistant/components/google_assistant_sdk/config_flow.py

@@ -0,0 +1,67 @@
+"""Config flow for Google Assistant SDK integration."""
+from __future__ import annotations
+
+from collections.abc import Mapping
+import logging
+from typing import Any
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers import config_entry_oauth2_flow
+
+from .const import DEFAULT_NAME, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class OAuth2FlowHandler(
+    config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
+):
+    """Config flow to handle Google Assistant SDK OAuth2 authentication."""
+
+    DOMAIN = DOMAIN
+
+    reauth_entry: ConfigEntry | None = None
+
+    @property
+    def logger(self) -> logging.Logger:
+        """Return logger."""
+        return logging.getLogger(__name__)
+
+    @property
+    def extra_authorize_data(self) -> dict[str, Any]:
+        """Extra data that needs to be appended to the authorize url."""
+        return {
+            "scope": "https://www.googleapis.com/auth/assistant-sdk-prototype",
+            # Add params to ensure we get back a refresh token
+            "access_type": "offline",
+            "prompt": "consent",
+        }
+
+    async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
+        """Perform reauth upon an API authentication error."""
+        self.reauth_entry = self.hass.config_entries.async_get_entry(
+            self.context["entry_id"]
+        )
+        return await self.async_step_reauth_confirm()
+
+    async def async_step_reauth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Confirm reauth dialog."""
+        if user_input is None:
+            return self.async_show_form(step_id="reauth_confirm")
+        return await self.async_step_user()
+
+    async def async_oauth_create_entry(self, data: dict[str, Any]) -> FlowResult:
+        """Create an entry for the flow, or update existing entry."""
+        if self.reauth_entry:
+            self.hass.config_entries.async_update_entry(self.reauth_entry, data=data)
+            await self.hass.config_entries.async_reload(self.reauth_entry.entry_id)
+            return self.async_abort(reason="reauth_successful")
+
+        if self._async_current_entries():
+            # Config entry already exists, only one allowed.
+            return self.async_abort(reason="single_instance_allowed")
+
+        return self.async_create_entry(title=DEFAULT_NAME, data=data)

homeassistant/components/google_assistant_sdk/const.py

@@ -0,0 +1,6 @@
+"""Constants for Google Assistant SDK integration."""
+from __future__ import annotations
+
+DOMAIN = "google_assistant_sdk"
+
+DEFAULT_NAME = "Google Assistant SDK"

homeassistant/components/google_assistant_sdk/helpers.py

@@ -0,0 +1,32 @@
+"""Helper classes for Google Assistant SDK integration."""
+from __future__ import annotations
+
+import aiohttp
+from gassist_text import TextAssistant
+from google.oauth2.credentials import Credentials
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_ACCESS_TOKEN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.config_entry_oauth2_flow import OAuth2Session
+
+from .const import DOMAIN
+
+
+async def async_send_text_commands(commands: list[str], hass: HomeAssistant) -> None:
+    """Send text commands to Google Assistant Service."""
+    # There can only be 1 entry (config_flow has single_instance_allowed)
+    entry: ConfigEntry = hass.config_entries.async_entries(DOMAIN)[0]
+
+    session: OAuth2Session = hass.data[DOMAIN].get(entry.entry_id)
+    try:
+        await session.async_ensure_token_valid()
+    except aiohttp.ClientResponseError as err:
+        if 400 <= err.status < 500:
+            entry.async_start_reauth(hass)
+        raise err
+
+    credentials = Credentials(session.token[CONF_ACCESS_TOKEN])
+    with TextAssistant(credentials) as assistant:
+        for command in commands:
+            assistant.assist(command)

homeassistant/components/google_assistant_sdk/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "google_assistant_sdk",
+  "name": "Google Assistant SDK",
+  "config_flow": true,
+  "dependencies": ["application_credentials"],
+  "documentation": "https://www.home-assistant.io/integrations/google_assistant_sdk/",
+  "requirements": ["gassist-text==0.0.4"],
+  "codeowners": ["@tronikos"],
+  "iot_class": "cloud_polling",
+  "integration_type": "service"
+}

homeassistant/components/google_assistant_sdk/notify.py

@@ -0,0 +1,41 @@
+"""Support for Google Assistant SDK broadcast notifications."""
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+
+from .helpers import async_send_text_commands
+
+
+async def async_get_service(
+    hass: HomeAssistant,
+    config: ConfigType,
+    discovery_info: DiscoveryInfoType | None = None,
+) -> BaseNotificationService:
+    """Get the broadcast notification service."""
+    return BroadcastNotificationService(hass)
+
+
+class BroadcastNotificationService(BaseNotificationService):
+    """Implement broadcast notification service."""
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize the service."""
+        self.hass = hass
+
+    async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
+        """Send a message."""
+        if not message:
+            return
+
+        commands = []
+        targets = kwargs.get(ATTR_TARGET)
+        if not targets:
+            commands.append(f"broadcast {message}")
+        else:
+            for target in targets:
+                commands.append(f"broadcast to {target} {message}")
+        await async_send_text_commands(commands, self.hass)

homeassistant/components/google_assistant_sdk/services.yaml

@@ -0,0 +1,10 @@
+send_text_command:
+  name: Send text command
+  description: Send a command as a text query to Google Assistant.
+  fields:
+    command:
+      name: Command
+      description: Command to send to Google Assistant.
+      example: turn off kitchen TV
+      selector:
+        text:

homeassistant/components/google_assistant_sdk/strings.json

@@ -0,0 +1,33 @@
+{
+  "config": {
+    "step": {
+      "pick_implementation": {
+        "title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
+      },
+      "auth": {
+        "title": "Link Google Account"
+      },
+      "reauth_confirm": {
+        "title": "[%key:common::config_flow::title::reauth%]",
+        "description": "The Google Assistant SDK integration needs to re-authenticate your account"
+      }
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
+      "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "timeout_connect": "[%key:common::config_flow::error::timeout_connect%]",
+      "oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
+      "missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
+      "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
+      "invalid_access_token": "[%key:common::config_flow::error::invalid_access_token%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "create_entry": {
+      "default": "[%key:common::config_flow::create_entry::authenticated%]"
+    }
+  },
+  "application_credentials": {
+    "description": "Follow the [instructions]({more_info_url}) for [OAuth consent screen]({oauth_consent_url}) to give Home Assistant access to your Google Assistant SDK. You also need to create Application Credentials linked to your account:\n1. Go to [Credentials]({oauth_creds_url}) and click **Create Credentials**.\n1. From the drop-down list select **OAuth client ID**.\n1. Select **Web application** for the Application Type.\n\n"
+  }
+}

homeassistant/components/google_assistant_sdk/translations/en.json

@@ -0,0 +1,33 @@
+{
+    "application_credentials": {
+        "description": "Follow the [instructions]({more_info_url}) for [OAuth consent screen]({oauth_consent_url}) to give Home Assistant access to your Google Assistant SDK. You also need to create Application Credentials linked to your account:\n1. Go to [Credentials]({oauth_creds_url}) and click **Create Credentials**.\n1. From the drop-down list select **OAuth client ID**.\n1. Select **Web application** for the Application Type.\n\n"
+    },
+    "config": {
+        "abort": {
+            "already_configured": "Account is already configured",
+            "already_in_progress": "Configuration flow is already in progress",
+            "cannot_connect": "Failed to connect",
+            "invalid_access_token": "Invalid access token",
+            "missing_configuration": "The component is not configured. Please follow the documentation.",
+            "oauth_error": "Received invalid token data.",
+            "reauth_successful": "Re-authentication was successful",
+            "timeout_connect": "Timeout establishing connection",
+            "unknown": "Unexpected error"
+        },
+        "create_entry": {
+            "default": "Successfully authenticated"
+        },
+        "step": {
+            "auth": {
+                "title": "Link Google Account"
+            },
+            "pick_implementation": {
+                "title": "Pick Authentication Method"
+            },
+            "reauth_confirm": {
+                "description": "The Google Assistant SDK integration needs to re-authenticate your account",
+                "title": "Reauthenticate Integration"
+            }
+        }
+    }
+}

homeassistant/generated/application_credentials.py

@@ -6,6 +6,7 @@ To update, run python3 -m script.hassfest
 APPLICATION_CREDENTIALS = [
     "geocaching",
     "google",
+    "google_assistant_sdk",
     "google_sheets",
     "home_connect",
     "lametric",

homeassistant/generated/config_flows.py

@@ -154,6 +154,7 @@ FLOWS = {
         "gogogate2",
         "goodwe",
         "google",
+        "google_assistant_sdk",
         "google_sheets",
         "google_travel_time",
         "govee_ble",

homeassistant/generated/integrations.json

@@ -1957,6 +1957,12 @@
           "iot_class": "cloud_push",
           "name": "Google Assistant"
         },
+        "google_assistant_sdk": {
+          "integration_type": "service",
+          "config_flow": true,
+          "iot_class": "cloud_polling",
+          "name": "Google Assistant SDK"
+        },
         "google_cloud": {
           "integration_type": "hub",
           "config_flow": false,

requirements_all.txt

@@ -734,6 +734,9 @@ fritzconnection==1.10.3
 # homeassistant.components.google_translate
 gTTS==2.2.4
 
+# homeassistant.components.google_assistant_sdk
+gassist-text==0.0.4
+
 # homeassistant.components.google
 gcal-sync==4.0.4
 

requirements_test_all.txt

@@ -550,6 +550,9 @@ fritzconnection==1.10.3
 # homeassistant.components.google_translate
 gTTS==2.2.4
 
+# homeassistant.components.google_assistant_sdk
+gassist-text==0.0.4
+
 # homeassistant.components.google
 gcal-sync==4.0.4
 

tests/components/google_assistant_sdk/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Google Assistant SDK integration."""

tests/components/google_assistant_sdk/conftest.py

@@ -0,0 +1,82 @@
+"""PyTest fixtures and test helpers."""
+from collections.abc import Awaitable, Callable, Generator
+import time
+
+import pytest
+
+from homeassistant.components.application_credentials import (
+    ClientCredential,
+    async_import_client_credential,
+)
+from homeassistant.components.google_assistant_sdk.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+from tests.common import MockConfigEntry
+
+ComponentSetup = Callable[[], Awaitable[None]]
+
+CLIENT_ID = "1234"
+CLIENT_SECRET = "5678"
+
+
+@pytest.fixture
+async def setup_credentials(hass: HomeAssistant) -> None:
+    """Fixture to setup credentials."""
+    assert await async_setup_component(hass, "application_credentials", {})
+    await async_import_client_credential(
+        hass,
+        DOMAIN,
+        ClientCredential(CLIENT_ID, CLIENT_SECRET),
+    )
+
+
+@pytest.fixture(name="scopes")
+def mock_scopes() -> list[str]:
+    """Fixture to set the scopes present in the OAuth token."""
+    return ["https://www.googleapis.com/auth/assistant-sdk-prototype"]
+
+
+@pytest.fixture(name="expires_at")
+def mock_expires_at() -> int:
+    """Fixture to set the oauth token expiration time."""
+    return time.time() + 3600
+
+
+@pytest.fixture(name="config_entry")
+def mock_config_entry(expires_at: int, scopes: list[str]) -> MockConfigEntry:
+    """Fixture for MockConfigEntry."""
+    return MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            "auth_implementation": DOMAIN,
+            "token": {
+                "access_token": "mock-access-token",
+                "refresh_token": "mock-refresh-token",
+                "expires_at": expires_at,
+                "scope": " ".join(scopes),
+            },
+        },
+    )
+
+
+@pytest.fixture(name="setup_integration")
+async def mock_setup_integration(
+    hass: HomeAssistant, config_entry: MockConfigEntry
+) -> Generator[ComponentSetup, None, None]:
+    """Fixture for setting up the component."""
+    config_entry.add_to_hass(hass)
+
+    assert await async_setup_component(hass, "application_credentials", {})
+    await async_import_client_credential(
+        hass,
+        DOMAIN,
+        ClientCredential("client-id", "client-secret"),
+        DOMAIN,
+    )
+
+    async def func() -> None:
+        assert await async_setup_component(hass, DOMAIN, {})
+        await hass.async_block_till_done()
+
+    yield func

tests/components/google_assistant_sdk/test_config_flow.py

@@ -0,0 +1,207 @@
+"""Test the Google Assistant SDK config flow."""
+from unittest.mock import patch
+
+import oauth2client
+
+from homeassistant import config_entries
+from homeassistant.components.google_assistant_sdk.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_entry_oauth2_flow
+
+from .conftest import CLIENT_ID
+
+from tests.common import MockConfigEntry
+
+TITLE = "Google Assistant SDK"
+
+
+async def test_full_flow(
+    hass: HomeAssistant,
+    hass_client_no_auth,
+    aioclient_mock,
+    current_request_with_host,
+    setup_credentials,
+) -> None:
+    """Check full flow."""
+    result = await hass.config_entries.flow.async_init(
+        "google_assistant_sdk", context={"source": config_entries.SOURCE_USER}
+    )
+    state = config_entry_oauth2_flow._encode_jwt(
+        hass,
+        {
+            "flow_id": result["flow_id"],
+            "redirect_uri": "https://example.com/auth/external/callback",
+        },
+    )
+
+    assert result["url"] == (
+        f"{oauth2client.GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
+        "&redirect_uri=https://example.com/auth/external/callback"
+        f"&state={state}&scope=https://www.googleapis.com/auth/assistant-sdk-prototype"
+        "&access_type=offline&prompt=consent"
+    )
+
+    client = await hass_client_no_auth()
+    resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
+    assert resp.status == 200
+    assert resp.headers["content-type"] == "text/html; charset=utf-8"
+
+    aioclient_mock.post(
+        oauth2client.GOOGLE_TOKEN_URI,
+        json={
+            "refresh_token": "mock-refresh-token",
+            "access_token": "mock-access-token",
+            "type": "Bearer",
+            "expires_in": 60,
+        },
+    )
+
+    with patch(
+        "homeassistant.components.google_assistant_sdk.async_setup_entry",
+        return_value=True,
+    ) as mock_setup:
+        result = await hass.config_entries.flow.async_configure(result["flow_id"])
+
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 1
+    assert len(mock_setup.mock_calls) == 1
+
+    assert result.get("type") == "create_entry"
+    assert result.get("title") == TITLE
+    assert "result" in result
+    assert result.get("result").unique_id is None
+    assert "token" in result.get("result").data
+    assert result.get("result").data["token"].get("access_token") == "mock-access-token"
+    assert (
+        result.get("result").data["token"].get("refresh_token") == "mock-refresh-token"
+    )
+
+
+async def test_reauth(
+    hass: HomeAssistant,
+    hass_client_no_auth,
+    aioclient_mock,
+    current_request_with_host,
+    setup_credentials,
+) -> None:
+    """Test the reauthentication case updates the existing config entry."""
+
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            "token": {
+                "access_token": "mock-access-token",
+            },
+        },
+    )
+    config_entry.add_to_hass(hass)
+
+    config_entry.async_start_reauth(hass)
+    await hass.async_block_till_done()
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    result = flows[0]
+    assert result["step_id"] == "reauth_confirm"
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+    state = config_entry_oauth2_flow._encode_jwt(
+        hass,
+        {
+            "flow_id": result["flow_id"],
+            "redirect_uri": "https://example.com/auth/external/callback",
+        },
+    )
+    assert result["url"] == (
+        f"{oauth2client.GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
+        "&redirect_uri=https://example.com/auth/external/callback"
+        f"&state={state}&scope=https://www.googleapis.com/auth/assistant-sdk-prototype"
+        "&access_type=offline&prompt=consent"
+    )
+    client = await hass_client_no_auth()
+    resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
+    assert resp.status == 200
+    assert resp.headers["content-type"] == "text/html; charset=utf-8"
+
+    aioclient_mock.post(
+        oauth2client.GOOGLE_TOKEN_URI,
+        json={
+            "refresh_token": "mock-refresh-token",
+            "access_token": "updated-access-token",
+            "type": "Bearer",
+            "expires_in": 60,
+        },
+    )
+
+    with patch(
+        "homeassistant.components.google_assistant_sdk.async_setup_entry",
+        return_value=True,
+    ) as mock_setup:
+        result = await hass.config_entries.flow.async_configure(result["flow_id"])
+
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 1
+    assert len(mock_setup.mock_calls) == 1
+
+    assert result.get("type") == "abort"
+    assert result.get("reason") == "reauth_successful"
+
+    assert config_entry.unique_id is None
+    assert "token" in config_entry.data
+    # Verify access token is refreshed
+    assert config_entry.data["token"].get("access_token") == "updated-access-token"
+    assert config_entry.data["token"].get("refresh_token") == "mock-refresh-token"
+
+
+async def test_single_instance_allowed(
+    hass: HomeAssistant,
+    hass_client_no_auth,
+    aioclient_mock,
+    current_request_with_host,
+    setup_credentials,
+) -> None:
+    """Test case where config flow allows a single test."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            "token": {
+                "access_token": "mock-access-token",
+            },
+        },
+    )
+    config_entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        "google_assistant_sdk", context={"source": config_entries.SOURCE_USER}
+    )
+    state = config_entry_oauth2_flow._encode_jwt(
+        hass,
+        {
+            "flow_id": result["flow_id"],
+            "redirect_uri": "https://example.com/auth/external/callback",
+        },
+    )
+
+    assert result["url"] == (
+        f"{oauth2client.GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
+        "&redirect_uri=https://example.com/auth/external/callback"
+        f"&state={state}&scope=https://www.googleapis.com/auth/assistant-sdk-prototype"
+        "&access_type=offline&prompt=consent"
+    )
+
+    client = await hass_client_no_auth()
+    resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
+    assert resp.status == 200
+    assert resp.headers["content-type"] == "text/html; charset=utf-8"
+
+    aioclient_mock.post(
+        oauth2client.GOOGLE_TOKEN_URI,
+        json={
+            "refresh_token": "mock-refresh-token",
+            "access_token": "mock-access-token",
+            "type": "Bearer",
+            "expires_in": 60,
+        },
+    )
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert result.get("type") == "abort"
+    assert result.get("reason") == "single_instance_allowed"

tests/components/google_assistant_sdk/test_init.py

@@ -0,0 +1,167 @@
+"""Tests for Google Assistant SDK."""
+import http
+import time
+from unittest.mock import patch
+
+import aiohttp
+import pytest
+
+from homeassistant.components.google_assistant_sdk import DOMAIN
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+
+from .conftest import ComponentSetup
+
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+
+async def test_setup_success(
+    hass: HomeAssistant, setup_integration: ComponentSetup
+) -> None:
+    """Test successful setup and unload."""
+    await setup_integration()
+
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert len(entries) == 1
+    assert entries[0].state is ConfigEntryState.LOADED
+
+    await hass.config_entries.async_unload(entries[0].entry_id)
+    await hass.async_block_till_done()
+
+    assert not hass.data.get(DOMAIN)
+    assert entries[0].state is ConfigEntryState.NOT_LOADED
+    assert not hass.services.async_services().get(DOMAIN, {})
+
+
+@pytest.mark.parametrize("expires_at", [time.time() - 3600], ids=["expired"])
+async def test_expired_token_refresh_success(
+    hass: HomeAssistant,
+    setup_integration: ComponentSetup,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test expired token is refreshed."""
+
+    aioclient_mock.post(
+        "https://oauth2.googleapis.com/token",
+        json={
+            "access_token": "updated-access-token",
+            "refresh_token": "updated-refresh-token",
+            "expires_at": time.time() + 3600,
+            "expires_in": 3600,
+        },
+    )
+
+    await setup_integration()
+
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert len(entries) == 1
+    assert entries[0].state is ConfigEntryState.LOADED
+    assert entries[0].data["token"]["access_token"] == "updated-access-token"
+    assert entries[0].data["token"]["expires_in"] == 3600
+
+
+@pytest.mark.parametrize(
+    "expires_at,status,expected_state",
+    [
+        (
+            time.time() - 3600,
+            http.HTTPStatus.UNAUTHORIZED,
+            ConfigEntryState.SETUP_ERROR,
+        ),
+        (
+            time.time() - 3600,
+            http.HTTPStatus.INTERNAL_SERVER_ERROR,
+            ConfigEntryState.SETUP_RETRY,
+        ),
+    ],
+    ids=["failure_requires_reauth", "transient_failure"],
+)
+async def test_expired_token_refresh_failure(
+    hass: HomeAssistant,
+    setup_integration: ComponentSetup,
+    aioclient_mock: AiohttpClientMocker,
+    status: http.HTTPStatus,
+    expected_state: ConfigEntryState,
+) -> None:
+    """Test failure while refreshing token with a transient error."""
+
+    aioclient_mock.post(
+        "https://oauth2.googleapis.com/token",
+        status=status,
+    )
+
+    await setup_integration()
+
+    # Verify a transient failure has occurred
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert entries[0].state is expected_state
+
+
+async def test_send_text_command(
+    hass: HomeAssistant,
+    setup_integration: ComponentSetup,
+) -> None:
+    """Test service call send_text_command calls TextAssistant."""
+    await setup_integration()
+
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert len(entries) == 1
+    assert entries[0].state is ConfigEntryState.LOADED
+
+    command = "turn on home assistant unsupported device"
+    with patch(
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist"
+    ) as mock_assist_call:
+        await hass.services.async_call(
+            DOMAIN,
+            "send_text_command",
+            {"command": command},
+            blocking=True,
+        )
+    mock_assist_call.assert_called_once_with(command)
+
+
+@pytest.mark.parametrize(
+    "status,requires_reauth",
+    [
+        (
+            http.HTTPStatus.UNAUTHORIZED,
+            True,
+        ),
+        (
+            http.HTTPStatus.INTERNAL_SERVER_ERROR,
+            False,
+        ),
+    ],
+    ids=["failure_requires_reauth", "transient_failure"],
+)
+async def test_send_text_command_expired_token_refresh_failure(
+    hass: HomeAssistant,
+    setup_integration: ComponentSetup,
+    aioclient_mock: AiohttpClientMocker,
+    status: http.HTTPStatus,
+    requires_reauth: ConfigEntryState,
+) -> None:
+    """Test failure refreshing token in send_text_command."""
+    await setup_integration()
+
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert len(entries) == 1
+    entry = entries[0]
+    assert entry.state is ConfigEntryState.LOADED
+
+    entry.data["token"]["expires_at"] = time.time() - 3600
+    aioclient_mock.post(
+        "https://oauth2.googleapis.com/token",
+        status=status,
+    )
+
+    with pytest.raises(aiohttp.ClientResponseError):
+        await hass.services.async_call(
+            DOMAIN,
+            "send_text_command",
+            {"command": "turn on tv"},
+            blocking=True,
+        )
+
+    assert any(entry.async_get_active_flows(hass, {"reauth"})) == requires_reauth

tests/components/google_assistant_sdk/test_notify.py

@@ -0,0 +1,92 @@
+"""Tests for the Google Assistant notify."""
+from unittest.mock import call, patch
+
+from homeassistant.components import notify
+from homeassistant.components.google_assistant_sdk import DOMAIN
+from homeassistant.core import HomeAssistant
+
+from .conftest import ComponentSetup
+
+
+async def test_broadcast_no_targets(
+    hass: HomeAssistant, setup_integration: ComponentSetup
+) -> None:
+    """Test broadcast to all."""
+    await setup_integration()
+
+    message = "time for dinner"
+    expected_command = "broadcast time for dinner"
+    with patch(
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist"
+    ) as mock_assist_call:
+        await hass.services.async_call(
+            notify.DOMAIN,
+            DOMAIN,
+            {notify.ATTR_MESSAGE: message},
+        )
+        await hass.async_block_till_done()
+    mock_assist_call.assert_called_once_with(expected_command)
+
+
+async def test_broadcast_one_target(
+    hass: HomeAssistant, setup_integration: ComponentSetup
+) -> None:
+    """Test broadcast to one target."""
+    await setup_integration()
+
+    message = "time for dinner"
+    target = "basement"
+    expected_command = "broadcast to basement time for dinner"
+    with patch(
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist"
+    ) as mock_assist_call:
+        await hass.services.async_call(
+            notify.DOMAIN,
+            DOMAIN,
+            {notify.ATTR_MESSAGE: message, notify.ATTR_TARGET: [target]},
+        )
+        await hass.async_block_till_done()
+    mock_assist_call.assert_called_once_with(expected_command)
+
+
+async def test_broadcast_two_targets(
+    hass: HomeAssistant, setup_integration: ComponentSetup
+) -> None:
+    """Test broadcast to two targets."""
+    await setup_integration()
+
+    message = "time for dinner"
+    target1 = "basement"
+    target2 = "master bedroom"
+    expected_command1 = "broadcast to basement time for dinner"
+    expected_command2 = "broadcast to master bedroom time for dinner"
+    with patch(
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist"
+    ) as mock_assist_call:
+        await hass.services.async_call(
+            notify.DOMAIN,
+            DOMAIN,
+            {notify.ATTR_MESSAGE: message, notify.ATTR_TARGET: [target1, target2]},
+        )
+        await hass.async_block_till_done()
+    mock_assist_call.assert_has_calls(
+        [call(expected_command1), call(expected_command2)]
+    )
+
+
+async def test_broadcast_empty_message(
+    hass: HomeAssistant, setup_integration: ComponentSetup
+) -> None:
+    """Test broadcast empty message."""
+    await setup_integration()
+
+    with patch(
+        "homeassistant.components.google_assistant_sdk.helpers.TextAssistant.assist"
+    ) as mock_assist_call:
+        await hass.services.async_call(
+            notify.DOMAIN,
+            DOMAIN,
+            {notify.ATTR_MESSAGE: ""},
+        )
+        await hass.async_block_till_done()
+    mock_assist_call.assert_not_called()