Add androidtv_remote.send_text action

homeassistant/components/androidtv_remote/__init__.py

@@ -5,26 +5,21 @@
 from asyncio import timeout
 import logging
 
-from androidtvremote2 import (
-    AndroidTVRemote,
-    CannotConnect,
-    ConnectionClosed,
-    InvalidAuth,
-)
+from androidtvremote2 import CannotConnect, ConnectionClosed, InvalidAuth
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_HOST, CONF_NAME, EVENT_HOMEASSISTANT_STOP, Platform
 from homeassistant.core import Event, HomeAssistant, callback
 from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
 
+from .const import AndroidTVRemoteConfigEntry
 from .helpers import create_api, get_enable_ime
+from .services import async_register_services
 
 _LOGGER = logging.getLogger(__name__)
 
 PLATFORMS: list[Platform] = [Platform.MEDIA_PLAYER, Platform.REMOTE]
 
-AndroidTVRemoteConfigEntry = ConfigEntry[AndroidTVRemote]
-
 
 async def async_setup_entry(
     hass: HomeAssistant, entry: AndroidTVRemoteConfigEntry
@@ -79,6 +74,8 @@ def on_hass_stop(event: Event) -> None:
     entry.async_on_unload(entry.add_update_listener(async_update_options))
     entry.async_on_unload(api.disconnect)
 
+    async_register_services(hass)
+
     return True
 
 

homeassistant/components/androidtv_remote/const.py

@@ -4,10 +4,16 @@
 
 from typing import Final
 
+from androidtvremote2 import AndroidTVRemote
+
+from homeassistant.config_entries import ConfigEntry
+
 DOMAIN: Final = "androidtv_remote"
 
 CONF_APPS = "apps"
 CONF_ENABLE_IME: Final = "enable_ime"
 CONF_ENABLE_IME_DEFAULT_VALUE: Final = True
 CONF_APP_NAME = "app_name"
 CONF_APP_ICON = "app_icon"
+
+AndroidTVRemoteConfigEntry = ConfigEntry[AndroidTVRemote]

homeassistant/components/androidtv_remote/icons.json

@@ -0,0 +1,7 @@
+{
+  "services": {
+    "send_text": {
+      "service": "mdi:keyboard"
+    }
+  }
+}

homeassistant/components/androidtv_remote/services.py

@@ -0,0 +1,60 @@
+"""Android TV Remote services."""
+
+from __future__ import annotations
+
+from androidtvremote2 import ConnectionClosed
+import voluptuous as vol
+
+from homeassistant.core import (
+    HomeAssistant,
+    ServiceCall,
+    ServiceResponse,
+    SupportsResponse,
+)
+from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+from homeassistant.helpers import config_validation as cv
+
+from .const import DOMAIN, AndroidTVRemoteConfigEntry
+
+CONF_CONFIG_ENTRY_ID = "config_entry_id"
+CONF_TEXT = "text"
+SEND_TEXT_SERVICE = "send_text"
+SEND_TEXT_SERVICE_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_CONFIG_ENTRY_ID): cv.string,
+        vol.Required(CONF_TEXT): cv.string,
+    }
+)
+
+
+def async_register_services(hass: HomeAssistant) -> None:
+    """Register Android TV Remote services."""
+
+    async def async_handle_send_text(call: ServiceCall) -> ServiceResponse:
+        """Send text."""
+        config_entry: AndroidTVRemoteConfigEntry | None = (
+            hass.config_entries.async_get_entry(call.data[CONF_CONFIG_ENTRY_ID])
+        )
+        if not config_entry:
+            raise ServiceValidationError(
+                translation_domain=DOMAIN,
+                translation_key="integration_not_found",
+                translation_placeholders={"target": DOMAIN},
+            )
+        api = config_entry.runtime_data
+        try:
+            api.send_text(call.data[CONF_TEXT])
+        except ConnectionClosed as exc:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN, translation_key="connection_closed"
+            ) from exc
+        return None
+
+    if not hass.services.has_service(DOMAIN, SEND_TEXT_SERVICE):
+        hass.services.async_register(
+            DOMAIN,
+            SEND_TEXT_SERVICE,
+            async_handle_send_text,
+            schema=SEND_TEXT_SERVICE_SCHEMA,
+            supports_response=SupportsResponse.NONE,
+        )

homeassistant/components/androidtv_remote/services.yaml

@@ -0,0 +1,12 @@
+send_text:
+  fields:
+    config_entry_id:
+      required: true
+      selector:
+        config_entry:
+          integration: androidtv_remote
+    text:
+      required: true
+      example: "hello world"
+      selector:
+        text:

homeassistant/components/androidtv_remote/strings.json

@@ -58,6 +58,25 @@
   "exceptions": {
     "connection_closed": {
       "message": "Connection to the Android TV device is closed"
+    },
+    "integration_not_found": {
+      "message": "Integration \"{target}\" not found in registry."
+    }
+  },
+  "services": {
+    "send_text": {
+      "name": "Send text",
+      "description": "Sends text to an Android TV device.",
+      "fields": {
+        "config_entry_id": {
+          "name": "Integration ID",
+          "description": "The Android TV Remote integration ID."
+        },
+        "text": {
+          "name": "Text",
+          "description": "Text to send as input to device."
+        }
+      }
     }
   }
 }

tests/components/androidtv_remote/test_services.py

@@ -0,0 +1,104 @@
+"""Tests for Android TV Remote services."""
+
+from unittest.mock import MagicMock, call
+
+from androidtvremote2 import ConnectionClosed
+import pytest
+
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+
+from tests.common import MockConfigEntry
+
+TEST_TEXT = "Hello World"
+
+
+async def setup_integration(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Set up the Android TV Remote integration for testing."""
+    mock_config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+    assert mock_config_entry.state is ConfigEntryState.LOADED
+    assert hass.services.has_service("androidtv_remote", "send_text")
+
+
+async def test_send_text_service(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
+) -> None:
+    """Test service call to send_text."""
+    await setup_integration(hass, mock_config_entry)
+    response = await hass.services.async_call(
+        "androidtv_remote",
+        "send_text",
+        {
+            "config_entry_id": mock_config_entry.entry_id,
+            "text": TEST_TEXT,
+        },
+        blocking=True,
+    )
+    assert response is None
+    assert mock_api.send_text.mock_calls == [call(TEST_TEXT)]
+
+
+async def test_send_text_service_config_entry_not_found(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
+) -> None:
+    """Test send_text service call with a config entry that does not exist."""
+    await setup_integration(hass, mock_config_entry)
+    with pytest.raises(HomeAssistantError, match="not found in registry"):
+        await hass.services.async_call(
+            "androidtv_remote",
+            "send_text",
+            {
+                "config_entry_id": "invalid-config-entry-id",
+                "text": TEST_TEXT,
+            },
+            blocking=True,
+        )
+
+
+async def test_config_entry_not_loaded(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
+) -> None:
+    """Test send_text service call with a config entry that is not loaded."""
+    await setup_integration(hass, mock_config_entry)
+    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
+
+    with pytest.raises(HomeAssistantError, match="not found in registry"):
+        await hass.services.async_call(
+            "androidtv_remote",
+            "send_text",
+            {
+                "config_entry_id": mock_config_entry.unique_id,
+                "text": TEST_TEXT,
+            },
+            blocking=True,
+        )
+
+
+async def test_send_text_service_fails(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
+) -> None:
+    """Test service call to send_text fails."""
+    await setup_integration(hass, mock_config_entry)
+    mock_api.send_text.side_effect = ConnectionClosed()
+
+    with pytest.raises(
+        HomeAssistantError, match="Connection to the Android TV device is closed"
+    ):
+        await hass.services.async_call(
+            "androidtv_remote",
+            "send_text",
+            {
+                "config_entry_id": mock_config_entry.entry_id,
+                "text": TEST_TEXT,
+            },
+            blocking=True,
+        )
+    assert mock_api.send_text.mock_calls == [call(TEST_TEXT)]