Enable local fulfillment google assistant (#63218)

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>

homeassistant/components/cloud/google_config.py

@@ -65,16 +65,11 @@ class CloudGoogleConfig(AbstractConfig):
         """Return if states should be proactively reported."""
         return self.enabled and self._prefs.google_report_state
 
-    @property
+    def get_local_webhook_id(self, agent_user_id):
-    def local_sdk_webhook_id(self):
+        """Return the webhook ID to be used for actions for a given agent user id via the local SDK."""
-        """Return the local SDK webhook.
-
-        Return None to disable the local SDK.
-        """
         return self._prefs.google_local_webhook_id
 
-    @property
+    def get_local_agent_user_id(self, webhook_id):
-    def local_sdk_user_id(self):
         """Return the user ID to be used for actions received via the local SDK."""
         return self._user
 

homeassistant/components/google_assistant/const.py

@@ -177,6 +177,7 @@ CHALLENGE_FAILED_PIN_NEEDED = "challengeFailedPinNeeded"
 CHALLENGE_PIN_NEEDED = "pinNeeded"
 
 STORE_AGENT_USER_IDS = "agent_user_ids"
+STORE_GOOGLE_LOCAL_WEBHOOK_ID = "local_webhook_id"
 
 SOURCE_CLOUD = "cloud"
 SOURCE_LOCAL = "local"

homeassistant/components/google_assistant/helpers.py

@@ -38,6 +38,7 @@ from .const import (
     NOT_EXPOSE_LOCAL,
     SOURCE_LOCAL,
     STORE_AGENT_USER_IDS,
+    STORE_GOOGLE_LOCAL_WEBHOOK_ID,
 )
 from .error import SmartHomeError
 
@@ -107,7 +108,7 @@ class AbstractConfig(ABC):
     async def async_initialize(self):
         """Perform async initialization of config."""
         self._store = GoogleConfigStore(self.hass)
-        await self._store.async_load()
+        await self._store.async_initialize()
 
         if not self.enabled:
             return
@@ -148,18 +149,22 @@ class AbstractConfig(ABC):
         """Return if states should be proactively reported."""
         return False
 
-    @property
+    def get_local_agent_user_id(self, webhook_id):
-    def local_sdk_webhook_id(self):
+        """Return the user ID to be used for actions received via the local SDK.
-        """Return the local SDK webhook ID.
 
-        Return None to disable the local SDK.
+        Return None is no agent user id is found.
         """
-        return None
+        found_agent_user_id = None
+        for agent_user_id, agent_user_data in self._store.agent_user_ids.items():
+            if agent_user_data[STORE_GOOGLE_LOCAL_WEBHOOK_ID] == webhook_id:
+                found_agent_user_id = agent_user_id
+                break
 
-    @property
+        return found_agent_user_id
-    def local_sdk_user_id(self):
+
-        """Return the user ID to be used for actions received via the local SDK."""
+    def get_local_webhook_id(self, agent_user_id):
-        raise NotImplementedError
+        """Return the webhook ID to be used for actions for a given agent user id via the local SDK."""
+        return self._store.agent_user_ids[agent_user_id][STORE_GOOGLE_LOCAL_WEBHOOK_ID]
 
     @abstractmethod
     def get_agent_user_id(self, context):
@@ -267,22 +272,41 @@ class AbstractConfig(ABC):
     @callback
     def async_enable_local_sdk(self):
         """Enable the local SDK."""
-        if (webhook_id := self.local_sdk_webhook_id) is None:
+        setup_successfull = True
-            return
+        setup_webhook_ids = []
+
+        for user_agent_id, _ in self._store.agent_user_ids.items():
+
+            if (webhook_id := self.get_local_webhook_id(user_agent_id)) is None:
+                setup_successfull = False
+                break
 
             try:
                 webhook.async_register(
                     self.hass,
                     DOMAIN,
-                "Local Support",
+                    "Local Support for " + user_agent_id,
                     webhook_id,
                     self._handle_local_webhook,
                 )
+                setup_webhook_ids.append(webhook_id)
             except ValueError:
-            _LOGGER.info("Webhook handler is already defined!")
+                _LOGGER.warning(
-            return
+                    "Webhook handler %s for agent user id %s is already defined!",
+                    webhook_id,
+                    user_agent_id,
+                )
+                setup_successfull = False
+                break
 
-        self._local_sdk_active = True
+        if not setup_successfull:
+            _LOGGER.warning(
+                "Local fulfillment failed to setup, falling back to cloud fulfillment"
+            )
+            for setup_webhook_id in setup_webhook_ids:
+                webhook.async_unregister(self.hass, setup_webhook_id)
+
+        self._local_sdk_active = setup_successfull
 
     @callback
     def async_disable_local_sdk(self):
@@ -290,7 +314,11 @@ class AbstractConfig(ABC):
         if not self._local_sdk_active:
             return
 
-        webhook.async_unregister(self.hass, self.local_sdk_webhook_id)
+        for agent_user_id in self._store.agent_user_ids:
+            webhook.async_unregister(
+                self.hass, self.get_local_webhook_id(agent_user_id)
+            )
+
         self._local_sdk_active = False
 
     async def _handle_local_webhook(self, hass, webhook_id, request):
@@ -307,8 +335,23 @@ class AbstractConfig(ABC):
         if not self.enabled:
             return json_response(smart_home.turned_off_response(payload))
 
+        if (agent_user_id := self.get_local_agent_user_id(webhook_id)) is None:
+            # No agent user linked to this webhook, means that the user has somehow unregistered
+            # removing webhook and stopping processing of this request.
+            _LOGGER.error(
+                "Cannot process request for webhook %s as no linked agent user is found:\n%s\n",
+                webhook_id,
+                pprint.pformat(payload),
+            )
+            webhook.async_unregister(self.hass, webhook_id)
+            return None
+
         result = await smart_home.async_handle_message(
-            self.hass, self, self.local_sdk_user_id, payload, SOURCE_LOCAL
+            self.hass,
+            self,
+            agent_user_id,
+            payload,
+            SOURCE_LOCAL,
         )
 
         if _LOGGER.isEnabledFor(logging.DEBUG):
@@ -327,7 +370,32 @@ class GoogleConfigStore:
         """Initialize a configuration store."""
         self._hass = hass
         self._store = Store(hass, self._STORAGE_VERSION, self._STORAGE_KEY)
-        self._data = {STORE_AGENT_USER_IDS: {}}
+        self._data = None
+
+    async def async_initialize(self):
+        """Finish initializing the ConfigStore."""
+        should_save_data = False
+        if (data := await self._store.async_load()) is None:
+            # if the store is not found create an empty one
+            # Note that the first request is always a cloud request,
+            # and that will store the correct agent user id to be used for local requests
+            data = {
+                STORE_AGENT_USER_IDS: {},
+            }
+            should_save_data = True
+
+        for agent_user_id, agent_user_data in data[STORE_AGENT_USER_IDS].items():
+            if STORE_GOOGLE_LOCAL_WEBHOOK_ID not in agent_user_data:
+                data[STORE_AGENT_USER_IDS][agent_user_id] = {
+                    **agent_user_data,
+                    STORE_GOOGLE_LOCAL_WEBHOOK_ID: webhook.async_generate_id(),
+                }
+                should_save_data = True
+
+        if should_save_data:
+            await self._store.async_save(data)
+
+        self._data = data
 
     @property
     def agent_user_ids(self):
@@ -338,7 +406,9 @@ class GoogleConfigStore:
     def add_agent_user_id(self, agent_user_id):
         """Add an agent user id to store."""
         if agent_user_id not in self._data[STORE_AGENT_USER_IDS]:
-            self._data[STORE_AGENT_USER_IDS][agent_user_id] = {}
+            self._data[STORE_AGENT_USER_IDS][agent_user_id] = {
+                STORE_GOOGLE_LOCAL_WEBHOOK_ID: webhook.async_generate_id(),
+            }
             self._store.async_delay_save(lambda: self._data, 1.0)
 
     @callback
@@ -348,11 +418,6 @@ class GoogleConfigStore:
             self._data[STORE_AGENT_USER_IDS].pop(agent_user_id, None)
             self._store.async_delay_save(lambda: self._data, 1.0)
 
-    async def async_load(self):
-        """Store current configuration to disk."""
-        if data := await self._store.async_load():
-            self._data = data
-
 
 class RequestData:
     """Hold data associated with a particular request."""
@@ -507,7 +572,7 @@ class GoogleEntity:
         if self.config.is_local_sdk_active and self.should_expose_local():
             device["otherDeviceIds"] = [{"deviceId": self.entity_id}]
             device["customData"] = {
-                "webhookId": self.config.local_sdk_webhook_id,
+                "webhookId": self.config.get_local_webhook_id(agent_user_id),
                 "httpPort": self.hass.http.server_port,
                 "httpSSL": self.hass.config.api.use_ssl,
                 "uuid": await self.hass.helpers.instance_id.async_get(),

homeassistant/components/google_assistant/http.py

@@ -12,9 +12,10 @@ from aiohttp import ClientError, ClientResponseError
 from aiohttp.web import Request, Response
 import jwt
 
-# Typing imports
 from homeassistant.components.http import HomeAssistantView
 from homeassistant.const import CLOUD_NEVER_EXPOSED_ENTITIES, ENTITY_CATEGORIES
+
+# Typing imports
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
@@ -84,6 +85,12 @@ class GoogleConfig(AbstractConfig):
         self._access_token = None
         self._access_token_renew = None
 
+    async def async_initialize(self):
+        """Perform async initialization of config."""
+        await super().async_initialize()
+
+        self.async_enable_local_sdk()
+
     @property
     def enabled(self):
         """Return if Google is enabled."""

homeassistant/components/google_assistant/smart_home.py

@@ -83,6 +83,8 @@ async def async_devices_sync(hass, data, payload):
     )
 
     agent_user_id = data.config.get_agent_user_id(data.context)
+    await data.config.async_connect_agent_user(agent_user_id)
+
     entities = async_get_entities(hass, data.config)
     results = await asyncio.gather(
         *(
@@ -103,8 +105,6 @@ async def async_devices_sync(hass, data, payload):
 
     response = {"agentUserId": agent_user_id, "devices": devices}
 
-    await data.config.async_connect_agent_user(agent_user_id)
-
     _LOGGER.debug("Syncing entities response: %s", response)
 
     return response

tests/components/google_assistant/__init__.py

@@ -24,8 +24,6 @@ class MockConfig(helpers.AbstractConfig):
         enabled=True,
         entity_config=None,
         hass=None,
-        local_sdk_user_id=None,
-        local_sdk_webhook_id=None,
         secure_devices_pin=None,
         should_2fa=None,
         should_expose=None,
@@ -35,8 +33,6 @@ class MockConfig(helpers.AbstractConfig):
         super().__init__(hass)
         self._enabled = enabled
         self._entity_config = entity_config or {}
-        self._local_sdk_user_id = local_sdk_user_id
-        self._local_sdk_webhook_id = local_sdk_webhook_id
         self._secure_devices_pin = secure_devices_pin
         self._should_2fa = should_2fa
         self._should_expose = should_expose
@@ -58,16 +54,6 @@ class MockConfig(helpers.AbstractConfig):
         """Return secure devices pin."""
         return self._entity_config
 
-    @property
-    def local_sdk_webhook_id(self):
-        """Return local SDK webhook id."""
-        return self._local_sdk_webhook_id
-
-    @property
-    def local_sdk_user_id(self):
-        """Return local SDK webhook id."""
-        return self._local_sdk_user_id
-
     def get_agent_user_id(self, context):
         """Get agent user ID making request."""
         return context.user_id

tests/components/google_assistant/test_helpers.py

@@ -9,6 +9,9 @@ from homeassistant.components.google_assistant import helpers
 from homeassistant.components.google_assistant.const import (
     EVENT_COMMAND_RECEIVED,
     NOT_EXPOSE_LOCAL,
+    SOURCE_CLOUD,
+    SOURCE_LOCAL,
+    STORE_GOOGLE_LOCAL_WEBHOOK_ID,
 )
 from homeassistant.config import async_process_ha_core_config
 from homeassistant.core import State
@@ -36,8 +39,11 @@ async def test_google_entity_sync_serialize_with_local_sdk(hass):
     hass.http = Mock(server_port=1234)
     config = MockConfig(
         hass=hass,
-        local_sdk_webhook_id="mock-webhook-id",
+        agent_user_ids={
-        local_sdk_user_id="mock-user-id",
+            "mock-user-id": {
+                STORE_GOOGLE_LOCAL_WEBHOOK_ID: "mock-webhook-id",
+            },
+        },
     )
     entity = helpers.GoogleEntity(hass, config, hass.states.get("light.ceiling_lights"))
 
@@ -48,12 +54,12 @@ async def test_google_entity_sync_serialize_with_local_sdk(hass):
     config.async_enable_local_sdk()
 
     with patch("homeassistant.helpers.instance_id.async_get", return_value="abcdef"):
-        serialized = await entity.sync_serialize(None)
+        serialized = await entity.sync_serialize("mock-user-id")
         assert serialized["otherDeviceIds"] == [{"deviceId": "light.ceiling_lights"}]
         assert serialized["customData"] == {
             "httpPort": 1234,
             "httpSSL": True,
-            "proxyDeviceId": None,
+            "proxyDeviceId": "mock-user-id",
             "webhookId": "mock-webhook-id",
             "baseUrl": "https://hostname:1234",
             "uuid": "abcdef",
@@ -79,8 +85,11 @@ async def test_config_local_sdk(hass, hass_client):
 
     config = MockConfig(
         hass=hass,
-        local_sdk_webhook_id="mock-webhook-id",
+        agent_user_ids={
-        local_sdk_user_id="mock-user-id",
+            "mock-user-id": {
+                STORE_GOOGLE_LOCAL_WEBHOOK_ID: "mock-webhook-id",
+            },
+        },
     )
 
     client = await hass_client()
@@ -118,7 +127,7 @@ async def test_config_local_sdk(hass, hass_client):
     assert result["requestId"] == "mock-req-id"
 
     assert len(command_events) == 1
-    assert command_events[0].context.user_id == config.local_sdk_user_id
+    assert command_events[0].context.user_id == "mock-user-id"
 
     assert len(turn_on_calls) == 1
     assert turn_on_calls[0].context is command_events[0].context
@@ -137,8 +146,11 @@ async def test_config_local_sdk_if_disabled(hass, hass_client):
 
     config = MockConfig(
         hass=hass,
-        local_sdk_webhook_id="mock-webhook-id",
+        agent_user_ids={
-        local_sdk_user_id="mock-user-id",
+            "mock-user-id": {
+                STORE_GOOGLE_LOCAL_WEBHOOK_ID: "mock-webhook-id",
+            },
+        },
         enabled=False,
     )
 
@@ -171,35 +183,61 @@ async def test_agent_user_id_storage(hass, hass_storage):
         "version": 1,
         "minor_version": 1,
         "key": "google_assistant",
-        "data": {"agent_user_ids": {"agent_1": {}}},
+        "data": {
+            "agent_user_ids": {
+                "agent_1": {
+                    "local_webhook_id": "test_webhook",
+                }
+            },
+        },
     }
 
     store = helpers.GoogleConfigStore(hass)
-    await store.async_load()
+    await store.async_initialize()
 
     assert hass_storage["google_assistant"] == {
         "version": 1,
         "minor_version": 1,
         "key": "google_assistant",
-        "data": {"agent_user_ids": {"agent_1": {}}},
+        "data": {
+            "agent_user_ids": {
+                "agent_1": {
+                    "local_webhook_id": "test_webhook",
+                }
+            },
+        },
     }
 
     async def _check_after_delay(data):
         async_fire_time_changed(hass, dt.utcnow() + timedelta(seconds=2))
         await hass.async_block_till_done()
 
-        assert hass_storage["google_assistant"] == {
+        assert (
+            list(hass_storage["google_assistant"]["data"]["agent_user_ids"].keys())
+            == data
+        )
+
+    store.add_agent_user_id("agent_2")
+    await _check_after_delay(["agent_1", "agent_2"])
+
+    store.pop_agent_user_id("agent_1")
+    await _check_after_delay(["agent_2"])
+
+    hass_storage["google_assistant"] = {
         "version": 1,
         "minor_version": 1,
         "key": "google_assistant",
-            "data": data,
+        "data": {
+            "agent_user_ids": {"agent_1": {}},
+        },
     }
+    store = helpers.GoogleConfigStore(hass)
+    await store.async_initialize()
 
-    store.add_agent_user_id("agent_2")
+    assert (
-    await _check_after_delay({"agent_user_ids": {"agent_1": {}, "agent_2": {}}})
+        STORE_GOOGLE_LOCAL_WEBHOOK_ID
-
+        in hass_storage["google_assistant"]["data"]["agent_user_ids"]["agent_1"]
-    store.pop_agent_user_id("agent_1")
+    )
-    await _check_after_delay({"agent_user_ids": {"agent_2": {}}})
 
 
 async def test_agent_user_id_connect():
@@ -254,3 +292,17 @@ def test_supported_features_string(caplog):
     )
     assert entity.is_supported() is False
     assert "Entity test.entity_id contains invalid supported_features value invalid"
+
+
+def test_request_data():
+    """Test request data properties."""
+    config = MockConfig()
+    data = helpers.RequestData(
+        config, "test_user", SOURCE_LOCAL, "test_request_id", None
+    )
+    assert data.is_local_request is True
+
+    data = helpers.RequestData(
+        config, "test_user", SOURCE_CLOUD, "test_request_id", None
+    )
+    assert data.is_local_request is False

tests/components/google_assistant/test_http.py

@@ -5,14 +5,23 @@ from unittest.mock import ANY, patch
 
 from homeassistant.components.google_assistant import GOOGLE_ASSISTANT_SCHEMA
 from homeassistant.components.google_assistant.const import (
+    DOMAIN,
+    EVENT_COMMAND_RECEIVED,
     HOMEGRAPH_TOKEN_URL,
     REPORT_STATE_BASE_URL,
+    STORE_AGENT_USER_IDS,
+    STORE_GOOGLE_LOCAL_WEBHOOK_ID,
 )
 from homeassistant.components.google_assistant.http import (
     GoogleConfig,
     _get_homegraph_jwt,
     _get_homegraph_token,
 )
+from homeassistant.const import CLOUD_NEVER_EXPOSED_ENTITIES
+from homeassistant.core import State
+from homeassistant.setup import async_setup_component
+
+from tests.common import async_capture_events, async_mock_service
 
 DUMMY_CONFIG = GOOGLE_ASSISTANT_SCHEMA(
     {
@@ -97,7 +106,7 @@ async def test_update_access_token(hass):
         mock_get_token.assert_called_once()
 
 
-async def test_call_homegraph_api(hass, aioclient_mock, hass_storage):
+async def test_call_homegraph_api(hass, aioclient_mock, hass_storage, caplog):
     """Test the function to call the homegraph api."""
     config = GoogleConfig(hass, DUMMY_CONFIG)
     await config.async_initialize()
@@ -164,3 +173,237 @@ async def test_report_state(hass, aioclient_mock, hass_storage):
             REPORT_STATE_BASE_URL,
             {"requestId": ANY, "agentUserId": agent_user_id, "payload": message},
         )
+
+
+async def test_google_config_local_fulfillment(hass, aioclient_mock, hass_storage):
+    """Test the google config for local fulfillment."""
+    agent_user_id = "user"
+    local_webhook_id = "webhook"
+
+    hass_storage["google_assistant"] = {
+        "version": 1,
+        "minor_version": 1,
+        "key": "google_assistant",
+        "data": {
+            "agent_user_ids": {
+                agent_user_id: {
+                    "local_webhook_id": local_webhook_id,
+                }
+            },
+        },
+    }
+
+    config = GoogleConfig(hass, DUMMY_CONFIG)
+    await config.async_initialize()
+
+    with patch.object(config, "async_call_homegraph_api"):
+        # Wait for google_assistant.helpers.async_initialize.sync_google to be called
+        await hass.async_block_till_done()
+
+    assert config.get_local_webhook_id(agent_user_id) == local_webhook_id
+    assert config.get_local_agent_user_id(local_webhook_id) == agent_user_id
+    assert config.get_local_agent_user_id("INCORRECT") is None
+
+
+async def test_secure_device_pin_config(hass):
+    """Test the setting of the secure device pin configuration."""
+    secure_pin = "TEST"
+    secure_config = GOOGLE_ASSISTANT_SCHEMA(
+        {
+            "project_id": "1234",
+            "service_account": {
+                "private_key": "-----BEGIN PRIVATE KEY-----\nMIICdwIBADANBgkqhkiG9w0BAQEFAASCAmEwggJdAgEAAoGBAKYscIlwm7soDsHAz6L6YvUkCvkrX19rS6yeYOmovvhoK5WeYGWUsd8V72zmsyHB7XO94YgJVjvxfzn5K8bLePjFzwoSJjZvhBJ/ZQ05d8VmbvgyWUoPdG9oEa4fZ/lCYrXoaFdTot2xcJvrb/ZuiRl4s4eZpNeFYvVK/Am7UeFPAgMBAAECgYAUetOfzLYUudofvPCaKHu7tKZ5kQPfEa0w6BAPnBF1Mfl1JiDBRDMryFtKs6AOIAVwx00dY/Ex0BCbB3+Cr58H7t4NaPTJxCpmR09pK7o17B7xAdQv8+SynFNud9/5vQ5AEXMOLNwKiU7wpXT6Z7ZIibUBOR7ewsWgsHCDpN1iqQJBAOMODPTPSiQMwRAUHIc6GPleFSJnIz2PAoG3JOG9KFAL6RtIc19lob2ZXdbQdzKtjSkWo+O5W20WDNAl1k32h6MCQQC7W4ZCIY67mPbL6CxXfHjpSGF4Dr9VWJ7ZrKHr6XUoOIcEvsn/pHvWonjMdy93rQMSfOE8BKd/I1+GHRmNVgplAkAnSo4paxmsZVyfeKt7Jy2dMY+8tVZe17maUuQaAE7Sk00SgJYegwrbMYgQnWCTL39HBfj0dmYA2Zj8CCAuu6O7AkEAryFiYjaUAO9+4iNoL27+ZrFtypeeadyov7gKs0ZKaQpNyzW8A+Zwi7TbTeSqzic/E+z/bOa82q7p/6b7141xsQJBANCAcIwMcVb6KVCHlQbOtKspo5Eh4ZQi8bGl+IcwbQ6JSxeTx915IfAldgbuU047wOB04dYCFB2yLDiUGVXTifU=\n-----END PRIVATE KEY-----\n",
+                "client_email": "dummy@dummy.iam.gserviceaccount.com",
+            },
+            "secure_devices_pin": secure_pin,
+        }
+    )
+    config = GoogleConfig(hass, secure_config)
+
+    assert config.secure_devices_pin == secure_pin
+
+
+async def test_should_expose(hass):
+    """Test the google config should expose method."""
+    config = GoogleConfig(hass, DUMMY_CONFIG)
+    await config.async_initialize()
+
+    with patch.object(config, "async_call_homegraph_api"):
+        # Wait for google_assistant.helpers.async_initialize.sync_google to be called
+        await hass.async_block_till_done()
+
+    assert (
+        config.should_expose(State(DOMAIN + ".mock", "mock", {"view": "not None"}))
+        is False
+    )
+
+    with patch.object(config, "async_call_homegraph_api"):
+        # Wait for google_assistant.helpers.async_initialize.sync_google to be called
+        await hass.async_block_till_done()
+
+    assert config.should_expose(State(CLOUD_NEVER_EXPOSED_ENTITIES[0], "mock")) is False
+
+
+async def test_missing_service_account(hass):
+    """Test the google config _async_request_sync_devices."""
+    incorrect_config = GOOGLE_ASSISTANT_SCHEMA(
+        {
+            "project_id": "1234",
+        }
+    )
+    config = GoogleConfig(hass, incorrect_config)
+    await config.async_initialize()
+
+    with patch.object(config, "async_call_homegraph_api"):
+        # Wait for google_assistant.helpers.async_initialize.sync_google to be called
+        await hass.async_block_till_done()
+
+    assert (
+        await config._async_request_sync_devices("mock")
+        is HTTPStatus.INTERNAL_SERVER_ERROR
+    )
+    renew = config._access_token_renew
+    await config._async_update_token()
+    assert config._access_token_renew is renew
+
+
+async def test_async_enable_local_sdk(hass, hass_client, hass_storage, caplog):
+    """Test the google config enable and disable local sdk."""
+    command_events = async_capture_events(hass, EVENT_COMMAND_RECEIVED)
+    turn_on_calls = async_mock_service(hass, "light", "turn_on")
+    hass.states.async_set("light.ceiling_lights", "off")
+
+    assert await async_setup_component(hass, "webhook", {})
+
+    hass_storage["google_assistant"] = {
+        "version": 1,
+        "minor_version": 1,
+        "key": "google_assistant",
+        "data": {
+            "agent_user_ids": {
+                "agent_1": {
+                    "local_webhook_id": "mock_webhook_id",
+                },
+            },
+        },
+    }
+    config = GoogleConfig(hass, DUMMY_CONFIG)
+    await config.async_initialize()
+
+    with patch.object(config, "async_call_homegraph_api"):
+        # Wait for google_assistant.helpers.async_initialize.sync_google to be called
+        await hass.async_block_till_done()
+
+    assert config.is_local_sdk_active is True
+
+    client = await hass_client()
+
+    resp = await client.post(
+        "/api/webhook/mock_webhook_id",
+        json={
+            "inputs": [
+                {
+                    "context": {"locale_country": "US", "locale_language": "en"},
+                    "intent": "action.devices.EXECUTE",
+                    "payload": {
+                        "commands": [
+                            {
+                                "devices": [{"id": "light.ceiling_lights"}],
+                                "execution": [
+                                    {
+                                        "command": "action.devices.commands.OnOff",
+                                        "params": {"on": True},
+                                    }
+                                ],
+                            }
+                        ],
+                        "structureData": {},
+                    },
+                }
+            ],
+            "requestId": "mock_req_id",
+        },
+    )
+    assert resp.status == HTTPStatus.OK
+    result = await resp.json()
+    assert result["requestId"] == "mock_req_id"
+
+    assert len(command_events) == 1
+    assert command_events[0].context.user_id == "agent_1"
+
+    assert len(turn_on_calls) == 1
+    assert turn_on_calls[0].context is command_events[0].context
+
+    config.async_disable_local_sdk()
+    assert config.is_local_sdk_active is False
+
+    config._store._data = {
+        STORE_AGENT_USER_IDS: {
+            "agent_1": {STORE_GOOGLE_LOCAL_WEBHOOK_ID: "mock_webhook_id"},
+            "agent_2": {STORE_GOOGLE_LOCAL_WEBHOOK_ID: "mock_webhook_id"},
+        },
+    }
+    config.async_enable_local_sdk()
+    assert config.is_local_sdk_active is False
+
+    config._store._data = {
+        STORE_AGENT_USER_IDS: {
+            "agent_1": {STORE_GOOGLE_LOCAL_WEBHOOK_ID: None},
+        },
+    }
+    config.async_enable_local_sdk()
+    assert config.is_local_sdk_active is False
+
+    config._store._data = {
+        STORE_AGENT_USER_IDS: {
+            "agent_2": {STORE_GOOGLE_LOCAL_WEBHOOK_ID: "mock_webhook_id"},
+            "agent_1": {STORE_GOOGLE_LOCAL_WEBHOOK_ID: None},
+        },
+    }
+    config.async_enable_local_sdk()
+    assert config.is_local_sdk_active is False
+
+    config.async_disable_local_sdk()
+
+    config._store._data = {
+        STORE_AGENT_USER_IDS: {
+            "agent_1": {STORE_GOOGLE_LOCAL_WEBHOOK_ID: "mock_webhook_id"},
+        },
+    }
+    config.async_enable_local_sdk()
+
+    config._store.pop_agent_user_id("agent_1")
+
+    caplog.clear()
+
+    resp = await client.post(
+        "/api/webhook/mock_webhook_id",
+        json={
+            "inputs": [
+                {
+                    "context": {"locale_country": "US", "locale_language": "en"},
+                    "intent": "action.devices.EXECUTE",
+                    "payload": {
+                        "commands": [
+                            {
+                                "devices": [{"id": "light.ceiling_lights"}],
+                                "execution": [
+                                    {
+                                        "command": "action.devices.commands.OnOff",
+                                        "params": {"on": True},
+                                    }
+                                ],
+                            }
+                        ],
+                        "structureData": {},
+                    },
+                }
+            ],
+            "requestId": "mock_req_id",
+        },
+    )
+    assert resp.status == HTTPStatus.OK
+    assert (
+        "Cannot process request for webhook mock_webhook_id as no linked agent user is found:"
+        in caplog.text
+    )

tests/components/google_assistant/test_smart_home.py

@@ -53,6 +53,58 @@ def registries(hass):
     return ret
 
 
+async def test_async_handle_message(hass):
+    """Test the async handle message method."""
+    config = MockConfig(
+        should_expose=lambda state: state.entity_id != "light.not_expose",
+        entity_config={
+            "light.demo_light": {
+                const.CONF_ROOM_HINT: "Living Room",
+                const.CONF_ALIASES: ["Hello", "World"],
+            }
+        },
+    )
+
+    result = await sh.async_handle_message(
+        hass,
+        config,
+        "test-agent",
+        {
+            "requestId": REQ_ID,
+            "inputs": [
+                {"intent": "action.devices.SYNC"},
+                {"intent": "action.devices.SYNC"},
+            ],
+        },
+        const.SOURCE_CLOUD,
+    )
+    assert result == {
+        "requestId": REQ_ID,
+        "payload": {"errorCode": const.ERR_PROTOCOL_ERROR},
+    }
+
+    await hass.async_block_till_done()
+
+    result = await sh.async_handle_message(
+        hass,
+        config,
+        "test-agent",
+        {
+            "requestId": REQ_ID,
+            "inputs": [
+                {"intent": "action.devices.DOES_NOT_EXIST"},
+            ],
+        },
+        const.SOURCE_CLOUD,
+    )
+    assert result == {
+        "requestId": REQ_ID,
+        "payload": {"errorCode": const.ERR_PROTOCOL_ERROR},
+    }
+
+    await hass.async_block_till_done()
+
+
 async def test_sync_message(hass):
     """Test a sync message."""
     light = DemoLight(