From f62fb76765513a5f1b0f96701fac29c1a35d636f Mon Sep 17 00:00:00 2001 From: Stephen Alderman Date: Tue, 16 Apr 2024 08:29:02 +0100 Subject: [PATCH] Add Config Flow to LG Netcast (#104913) * Add Config Flow to lg_netcast * Add YAML import to Lg Netcast ConfigFlow Deprecates YAML config support * Add LG Netcast Device triggers for turn_on action * Add myself to LG Netcast codeowners * Remove unnecessary user_input validation check. * Move netcast discovery logic to the backend * Use FlowResultType Enum for tests * Mock pylgnetcast.query_device_info instead of _send_to_tv * Refactor lg_netcast client discovery, simplify YAML import * Simplify CONF_NAME to use friendly name Fix: Use Friendly name for Name * Expose model to DeviceInfo * Add test for testing YAML import when not TV not online * Switch to entity_name for LGTVDevice * Add data_description to host field in user step * Wrap try only around _get_session_id * Send regular request for access_token to ensure it display on the TV * Stop displaying access token when flow is aborted * Remove config_flow only consts and minor fixups * Simplify media_player logic & raise new migration issue * Add async_unload_entry * Create issues when import config flow fails, and raise only a single yaml deprecation issue type * Remove single use trigger helpers * Bump issue deprecation breakage version * Lint --------- Co-authored-by: Erik Montnemery --- CODEOWNERS | 3 +- .../components/lg_netcast/__init__.py | 32 +++ .../components/lg_netcast/config_flow.py | 217 +++++++++++++++ homeassistant/components/lg_netcast/const.py | 6 + .../components/lg_netcast/device_trigger.py | 88 ++++++ .../components/lg_netcast/helpers.py | 59 ++++ .../components/lg_netcast/manifest.json | 7 +- .../components/lg_netcast/media_player.py | 92 +++++-- .../components/lg_netcast/strings.json | 46 ++++ .../components/lg_netcast/trigger.py | 49 ++++ .../lg_netcast/triggers/__init__.py | 1 + .../components/lg_netcast/triggers/turn_on.py | 115 ++++++++ homeassistant/generated/config_flows.py | 1 + homeassistant/generated/integrations.json | 4 +- requirements_all.txt | 2 +- requirements_test_all.txt | 3 + tests/components/lg_netcast/__init__.py | 116 ++++++++ tests/components/lg_netcast/conftest.py | 11 + .../components/lg_netcast/test_config_flow.py | 252 ++++++++++++++++++ .../lg_netcast/test_device_trigger.py | 148 ++++++++++ tests/components/lg_netcast/test_trigger.py | 189 +++++++++++++ 21 files changed, 1411 insertions(+), 30 deletions(-) create mode 100644 homeassistant/components/lg_netcast/config_flow.py create mode 100644 homeassistant/components/lg_netcast/device_trigger.py create mode 100644 homeassistant/components/lg_netcast/helpers.py create mode 100644 homeassistant/components/lg_netcast/strings.json create mode 100644 homeassistant/components/lg_netcast/trigger.py create mode 100644 homeassistant/components/lg_netcast/triggers/__init__.py create mode 100644 homeassistant/components/lg_netcast/triggers/turn_on.py create mode 100644 tests/components/lg_netcast/__init__.py create mode 100644 tests/components/lg_netcast/conftest.py create mode 100644 tests/components/lg_netcast/test_config_flow.py create mode 100644 tests/components/lg_netcast/test_device_trigger.py create mode 100644 tests/components/lg_netcast/test_trigger.py diff --git a/CODEOWNERS b/CODEOWNERS index 39fa804314d9ad..d93a8f6b9d36fb 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -753,7 +753,8 @@ build.json @home-assistant/supervisor /tests/components/leaone/ @bdraco /homeassistant/components/led_ble/ @bdraco /tests/components/led_ble/ @bdraco -/homeassistant/components/lg_netcast/ @Drafteed +/homeassistant/components/lg_netcast/ @Drafteed @splinter98 +/tests/components/lg_netcast/ @Drafteed @splinter98 /homeassistant/components/lidarr/ @tkdrob /tests/components/lidarr/ @tkdrob /homeassistant/components/light/ @home-assistant/core diff --git a/homeassistant/components/lg_netcast/__init__.py b/homeassistant/components/lg_netcast/__init__.py index 232d7bd10b85ce..f6fb834ab114f0 100644 --- a/homeassistant/components/lg_netcast/__init__.py +++ b/homeassistant/components/lg_netcast/__init__.py @@ -1 +1,33 @@ """The lg_netcast component.""" + +from typing import Final + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant +from homeassistant.helpers import config_validation as cv + +from .const import DOMAIN + +PLATFORMS: Final[list[Platform]] = [Platform.MEDIA_PLAYER] + +CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) + + +async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: + """Set up a config entry.""" + hass.data.setdefault(DOMAIN, {}) + + await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS) + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Unload a config entry.""" + unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + + if unload_ok: + del hass.data[DOMAIN][entry.entry_id] + + return unload_ok diff --git a/homeassistant/components/lg_netcast/config_flow.py b/homeassistant/components/lg_netcast/config_flow.py new file mode 100644 index 00000000000000..3c1d3d73e0f864 --- /dev/null +++ b/homeassistant/components/lg_netcast/config_flow.py @@ -0,0 +1,217 @@ +"""Config flow to configure the LG Netcast TV integration.""" + +from __future__ import annotations + +import contextlib +from datetime import datetime, timedelta +from typing import TYPE_CHECKING, Any + +from pylgnetcast import AccessTokenError, LgNetCastClient, SessionIdError +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.config_entries import ConfigFlowResult +from homeassistant.const import ( + CONF_ACCESS_TOKEN, + CONF_HOST, + CONF_ID, + CONF_MODEL, + CONF_NAME, +) +from homeassistant.core import CALLBACK_TYPE, DOMAIN as HOMEASSISTANT_DOMAIN, callback +from homeassistant.data_entry_flow import AbortFlow +from homeassistant.helpers.event import async_track_time_interval +from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue +from homeassistant.util.network import is_host_valid + +from .const import DEFAULT_NAME, DOMAIN +from .helpers import LGNetCastDetailDiscoveryError, async_discover_netcast_details + +DISPLAY_ACCESS_TOKEN_INTERVAL = timedelta(seconds=1) + + +class LGNetCast(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for LG Netcast TV integration.""" + + VERSION = 1 + + def __init__(self) -> None: + """Initialize config flow.""" + self.client: LgNetCastClient | None = None + self.device_config: dict[str, Any] = {} + self._discovered_devices: dict[str, Any] = {} + self._track_interval: CALLBACK_TYPE | None = None + + def create_client(self) -> None: + """Create LG Netcast client from config.""" + host = self.device_config[CONF_HOST] + access_token = self.device_config.get(CONF_ACCESS_TOKEN) + self.client = LgNetCastClient(host, access_token) + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle the initial step.""" + errors: dict[str, str] = {} + + if user_input is not None: + host = user_input[CONF_HOST] + if is_host_valid(host): + self.device_config[CONF_HOST] = host + return await self.async_step_authorize() + + errors[CONF_HOST] = "invalid_host" + + return self.async_show_form( + step_id="user", + data_schema=vol.Schema({vol.Required(CONF_HOST): str}), + errors=errors, + ) + + async def async_step_import(self, config: dict[str, Any]) -> ConfigFlowResult: + """Import configuration from yaml.""" + self.device_config = { + CONF_HOST: config[CONF_HOST], + CONF_NAME: config[CONF_NAME], + } + + def _create_issue(): + async_create_issue( + self.hass, + HOMEASSISTANT_DOMAIN, + f"deprecated_yaml_{DOMAIN}", + breaks_in_ha_version="2024.11.0", + is_fixable=False, + issue_domain=DOMAIN, + severity=IssueSeverity.WARNING, + translation_key="deprecated_yaml", + translation_placeholders={ + "domain": DOMAIN, + "integration_title": "LG Netcast", + }, + ) + + try: + result: ConfigFlowResult = await self.async_step_authorize(config) + except AbortFlow as err: + if err.reason != "already_configured": + async_create_issue( + self.hass, + DOMAIN, + "deprecated_yaml_import_issue_{err.reason}", + breaks_in_ha_version="2024.11.0", + is_fixable=False, + issue_domain=DOMAIN, + severity=IssueSeverity.WARNING, + translation_key=f"deprecated_yaml_import_issue_{err.reason}", + translation_placeholders={ + "domain": DOMAIN, + "integration_title": "LG Netcast", + "error_type": err.reason, + }, + ) + else: + _create_issue() + raise + + _create_issue() + + return result + + async def async_discover_client(self): + """Handle Discovery step.""" + self.create_client() + + if TYPE_CHECKING: + assert self.client is not None + + if self.device_config.get(CONF_ID): + return + + try: + details = await async_discover_netcast_details(self.hass, self.client) + except LGNetCastDetailDiscoveryError as err: + raise AbortFlow("cannot_connect") from err + + if (unique_id := details["uuid"]) is None: + raise AbortFlow("invalid_host") + + self.device_config[CONF_ID] = unique_id + self.device_config[CONF_MODEL] = details["model_name"] + + if CONF_NAME not in self.device_config: + self.device_config[CONF_NAME] = details["friendly_name"] or DEFAULT_NAME + + async def async_step_authorize( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle Authorize step.""" + errors: dict[str, str] = {} + self.async_stop_display_access_token() + + if user_input is not None and user_input.get(CONF_ACCESS_TOKEN) is not None: + self.device_config[CONF_ACCESS_TOKEN] = user_input[CONF_ACCESS_TOKEN] + + await self.async_discover_client() + assert self.client is not None + + await self.async_set_unique_id(self.device_config[CONF_ID]) + self._abort_if_unique_id_configured( + updates={CONF_HOST: self.device_config[CONF_HOST]} + ) + + try: + await self.hass.async_add_executor_job( + self.client._get_session_id # pylint: disable=protected-access + ) + except AccessTokenError: + if user_input is not None: + errors[CONF_ACCESS_TOKEN] = "invalid_access_token" + except SessionIdError: + errors["base"] = "cannot_connect" + else: + return await self.async_create_device() + + self._track_interval = async_track_time_interval( + self.hass, + self.async_display_access_token, + DISPLAY_ACCESS_TOKEN_INTERVAL, + cancel_on_shutdown=True, + ) + + return self.async_show_form( + step_id="authorize", + data_schema=vol.Schema( + { + vol.Optional(CONF_ACCESS_TOKEN): vol.All(str, vol.Length(max=6)), + } + ), + errors=errors, + ) + + async def async_display_access_token(self, _: datetime | None = None): + """Display access token on screen.""" + assert self.client is not None + with contextlib.suppress(AccessTokenError, SessionIdError): + await self.hass.async_add_executor_job( + self.client._get_session_id # pylint: disable=protected-access + ) + + @callback + def async_remove(self): + """Terminate Access token display if flow is removed.""" + self.async_stop_display_access_token() + + def async_stop_display_access_token(self): + """Stop Access token request if running.""" + if self._track_interval is not None: + self._track_interval() + self._track_interval = None + + async def async_create_device(self) -> ConfigFlowResult: + """Create LG Netcast TV Device from config.""" + assert self.client + + return self.async_create_entry( + title=self.device_config[CONF_NAME], data=self.device_config + ) diff --git a/homeassistant/components/lg_netcast/const.py b/homeassistant/components/lg_netcast/const.py index 0344ad6f177dfd..aca01c9b8708a1 100644 --- a/homeassistant/components/lg_netcast/const.py +++ b/homeassistant/components/lg_netcast/const.py @@ -1,3 +1,9 @@ """Constants for the lg_netcast component.""" +from typing import Final + +ATTR_MANUFACTURER: Final = "LG" + +DEFAULT_NAME: Final = "LG Netcast TV" + DOMAIN = "lg_netcast" diff --git a/homeassistant/components/lg_netcast/device_trigger.py b/homeassistant/components/lg_netcast/device_trigger.py new file mode 100644 index 00000000000000..51c5ec53004e8f --- /dev/null +++ b/homeassistant/components/lg_netcast/device_trigger.py @@ -0,0 +1,88 @@ +"""Provides device triggers for LG Netcast.""" + +from __future__ import annotations + +from typing import Any + +import voluptuous as vol + +from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA +from homeassistant.components.device_automation.exceptions import ( + InvalidDeviceAutomationConfig, +) +from homeassistant.const import CONF_DEVICE_ID, CONF_PLATFORM, CONF_TYPE +from homeassistant.core import CALLBACK_TYPE, HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo +from homeassistant.helpers.typing import ConfigType + +from . import trigger +from .const import DOMAIN +from .helpers import async_get_device_entry_by_device_id +from .triggers.turn_on import ( + PLATFORM_TYPE as TURN_ON_PLATFORM_TYPE, + async_get_turn_on_trigger, +) + +TRIGGER_TYPES = {TURN_ON_PLATFORM_TYPE} + +TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend( + { + vol.Required(CONF_TYPE): vol.In(TRIGGER_TYPES), + } +) + + +async def async_validate_trigger_config( + hass: HomeAssistant, config: ConfigType +) -> ConfigType: + """Validate config.""" + config = TRIGGER_SCHEMA(config) + + if config[CONF_TYPE] == TURN_ON_PLATFORM_TYPE: + device_id = config[CONF_DEVICE_ID] + + try: + device = async_get_device_entry_by_device_id(hass, device_id) + except ValueError as err: + raise InvalidDeviceAutomationConfig(err) from err + + if DOMAIN in hass.data: + for config_entry_id in device.config_entries: + if hass.data[DOMAIN].get(config_entry_id): + break + else: + raise InvalidDeviceAutomationConfig( + f"Device {device.id} is not from an existing {DOMAIN} config entry" + ) + + return config + + +async def async_get_triggers( + _hass: HomeAssistant, device_id: str +) -> list[dict[str, Any]]: + """List device triggers for LG Netcast devices.""" + return [async_get_turn_on_trigger(device_id)] + + +async def async_attach_trigger( + hass: HomeAssistant, + config: ConfigType, + action: TriggerActionType, + trigger_info: TriggerInfo, +) -> CALLBACK_TYPE: + """Attach a trigger.""" + if (trigger_type := config[CONF_TYPE]) == TURN_ON_PLATFORM_TYPE: + trigger_config = { + CONF_PLATFORM: trigger_type, + CONF_DEVICE_ID: config[CONF_DEVICE_ID], + } + trigger_config = await trigger.async_validate_trigger_config( + hass, trigger_config + ) + return await trigger.async_attach_trigger( + hass, trigger_config, action, trigger_info + ) + + raise HomeAssistantError(f"Unhandled trigger type {trigger_type}") diff --git a/homeassistant/components/lg_netcast/helpers.py b/homeassistant/components/lg_netcast/helpers.py new file mode 100644 index 00000000000000..7cfc0d502716b0 --- /dev/null +++ b/homeassistant/components/lg_netcast/helpers.py @@ -0,0 +1,59 @@ +"""Helper functions for LG Netcast TV.""" + +from typing import TypedDict +import xml.etree.ElementTree as ET + +from pylgnetcast import LgNetCastClient +from requests import RequestException + +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import device_registry as dr +from homeassistant.helpers.device_registry import DeviceEntry + +from .const import DOMAIN + + +class LGNetCastDetailDiscoveryError(Exception): + """Unable to retrieve details from Netcast TV.""" + + +class NetcastDetails(TypedDict): + """Netcast TV Details.""" + + uuid: str + model_name: str + friendly_name: str + + +async def async_discover_netcast_details( + hass: HomeAssistant, client: LgNetCastClient +) -> NetcastDetails: + """Discover UUID and Model Name from Netcast Tv.""" + try: + resp = await hass.async_add_executor_job(client.query_device_info) + except RequestException as err: + raise LGNetCastDetailDiscoveryError( + f"Error in connecting to {client.url}" + ) from err + except ET.ParseError as err: + raise LGNetCastDetailDiscoveryError("Invalid XML") from err + + if resp is None: + raise LGNetCastDetailDiscoveryError("Empty response received") + + return resp + + +@callback +def async_get_device_entry_by_device_id( + hass: HomeAssistant, device_id: str +) -> DeviceEntry: + """Get Device Entry from Device Registry by device ID. + + Raises ValueError if device ID is invalid. + """ + device_reg = dr.async_get(hass) + if (device := device_reg.async_get(device_id)) is None: + raise ValueError(f"Device {device_id} is not a valid {DOMAIN} device.") + + return device diff --git a/homeassistant/components/lg_netcast/manifest.json b/homeassistant/components/lg_netcast/manifest.json index 8a63e064b415ab..cf91374feb72d2 100644 --- a/homeassistant/components/lg_netcast/manifest.json +++ b/homeassistant/components/lg_netcast/manifest.json @@ -1,9 +1,12 @@ { "domain": "lg_netcast", "name": "LG Netcast", - "codeowners": ["@Drafteed"], + "codeowners": ["@Drafteed", "@splinter98"], + "config_flow": true, + "dependencies": ["network"], "documentation": "https://www.home-assistant.io/integrations/lg_netcast", + "integration_type": "device", "iot_class": "local_polling", "loggers": ["pylgnetcast"], - "requirements": ["pylgnetcast==0.3.7"] + "requirements": ["pylgnetcast==0.3.9"] } diff --git a/homeassistant/components/lg_netcast/media_player.py b/homeassistant/components/lg_netcast/media_player.py index 9f6e88dc614858..3fc07cab12bfbd 100644 --- a/homeassistant/components/lg_netcast/media_player.py +++ b/homeassistant/components/lg_netcast/media_player.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime -from typing import Any +from typing import TYPE_CHECKING, Any from pylgnetcast import LG_COMMAND, LgNetCastClient, LgNetCastError from requests import RequestException @@ -17,14 +17,19 @@ MediaPlayerState, MediaType, ) -from homeassistant.const import CONF_ACCESS_TOKEN, CONF_HOST, CONF_NAME +from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry +from homeassistant.const import CONF_ACCESS_TOKEN, CONF_HOST, CONF_MODEL, CONF_NAME from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType +from homeassistant.exceptions import PlatformNotReady import homeassistant.helpers.config_validation as cv +from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback -from homeassistant.helpers.script import Script +from homeassistant.helpers.trigger import PluggableAction from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType -from .const import DOMAIN +from .const import ATTR_MANUFACTURER, DOMAIN +from .triggers.turn_on import async_get_turn_on_trigger DEFAULT_NAME = "LG TV Remote" @@ -54,23 +59,45 @@ ) -def setup_platform( +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up a LG Netcast Media Player from a config_entry.""" + + host = config_entry.data[CONF_HOST] + access_token = config_entry.data[CONF_ACCESS_TOKEN] + unique_id = config_entry.unique_id + name = config_entry.data.get(CONF_NAME, DEFAULT_NAME) + model = config_entry.data[CONF_MODEL] + + client = LgNetCastClient(host, access_token) + + hass.data[DOMAIN][config_entry.entry_id] = client + + async_add_entities([LgTVDevice(client, name, model, unique_id=unique_id)]) + + +async def async_setup_platform( hass: HomeAssistant, config: ConfigType, - add_entities: AddEntitiesCallback, + async_add_entities: AddEntitiesCallback, discovery_info: DiscoveryInfoType | None = None, ) -> None: """Set up the LG TV platform.""" host = config.get(CONF_HOST) - access_token = config.get(CONF_ACCESS_TOKEN) - name = config[CONF_NAME] - on_action = config.get(CONF_ON_ACTION) - client = LgNetCastClient(host, access_token) - on_action_script = Script(hass, on_action, name, DOMAIN) if on_action else None + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_IMPORT}, data=config + ) - add_entities([LgTVDevice(client, name, on_action_script)], True) + if ( + result.get("type") == FlowResultType.ABORT + and result.get("reason") == "cannot_connect" + ): + raise PlatformNotReady(f"Connection error while connecting to {host}") class LgTVDevice(MediaPlayerEntity): @@ -79,19 +106,42 @@ class LgTVDevice(MediaPlayerEntity): _attr_assumed_state = True _attr_device_class = MediaPlayerDeviceClass.TV _attr_media_content_type = MediaType.CHANNEL + _attr_has_entity_name = True + _attr_name = None - def __init__(self, client, name, on_action_script): + def __init__(self, client, name, model, unique_id): """Initialize the LG TV device.""" self._client = client - self._name = name self._muted = False - self._on_action_script = on_action_script + self._turn_on = PluggableAction(self.async_write_ha_state) self._volume = 0 self._channel_id = None self._channel_name = "" self._program_name = "" self._sources = {} self._source_names = [] + self._attr_unique_id = unique_id + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, unique_id)}, + manufacturer=ATTR_MANUFACTURER, + name=name, + model=model, + ) + + async def async_added_to_hass(self) -> None: + """Connect and subscribe to dispatcher signals and state updates.""" + await super().async_added_to_hass() + + entry = self.registry_entry + + if TYPE_CHECKING: + assert entry is not None and entry.device_id is not None + + self.async_on_remove( + self._turn_on.async_register( + self.hass, async_get_turn_on_trigger(entry.device_id) + ) + ) def send_command(self, command): """Send remote control commands to the TV.""" @@ -151,11 +201,6 @@ def __update_volume(self): self._volume = volume self._muted = muted - @property - def name(self): - """Return the name of the device.""" - return self._name - @property def is_volume_muted(self): """Boolean if volume is currently muted.""" @@ -194,7 +239,7 @@ def media_title(self): @property def supported_features(self) -> MediaPlayerEntityFeature: """Flag media player features that are supported.""" - if self._on_action_script: + if self._turn_on: return SUPPORT_LGTV | MediaPlayerEntityFeature.TURN_ON return SUPPORT_LGTV @@ -209,10 +254,9 @@ def turn_off(self) -> None: """Turn off media player.""" self.send_command(LG_COMMAND.POWER) - def turn_on(self) -> None: + async def async_turn_on(self) -> None: """Turn on the media player.""" - if self._on_action_script: - self._on_action_script.run(context=self._context) + await self._turn_on.async_run(self.hass, self._context) def volume_up(self) -> None: """Volume up the media player.""" diff --git a/homeassistant/components/lg_netcast/strings.json b/homeassistant/components/lg_netcast/strings.json new file mode 100644 index 00000000000000..77003f60f43d5a --- /dev/null +++ b/homeassistant/components/lg_netcast/strings.json @@ -0,0 +1,46 @@ +{ + "config": { + "step": { + "user": { + "description": "Ensure that your TV is turned on before trying to set it up.\nIf you leave the host empty, discovery will be used to find devices.", + "data": { + "host": "[%key:common::config_flow::data::host%]" + }, + "data_description": { + "host": "The hostname or IP address of the LG Netcast TV to control." + } + }, + "authorize": { + "title": "Authorize LG Netcast TV", + "description": "Enter the Pairing Key displayed on the TV", + "data": { + "access_token": "[%key:common::config_flow::data::access_token%]" + } + } + }, + "error": { + "invalid_host": "[%key:common::config_flow::error::invalid_host%]", + "invalid_access_token": "[%key:common::config_flow::error::invalid_access_token%]", + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]" + }, + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]" + } + }, + "issues": { + "deprecated_yaml_import_issue_cannot_connect": { + "title": "The {integration_title} is not online for YAML migration to complete", + "description": "Migrating {integration_title} from YAML cannot complete until the TV is online.\n\nPlease turn on your TV for migration to complete." + }, + "deprecated_yaml_import_issue_invalid_host": { + "title": "The {integration_title} YAML configuration has an invalid host.", + "description": "Configuring {integration_title} using YAML is being removed but the device returned an invalid response.\n\nPlease check or manually remove the YAML configuration." + } + }, + "device_automation": { + "trigger_type": { + "lg_netcast.turn_on": "Device is requested to turn on" + } + } +} diff --git a/homeassistant/components/lg_netcast/trigger.py b/homeassistant/components/lg_netcast/trigger.py new file mode 100644 index 00000000000000..8dfbe309e038c2 --- /dev/null +++ b/homeassistant/components/lg_netcast/trigger.py @@ -0,0 +1,49 @@ +"""LG Netcast TV trigger dispatcher.""" + +from __future__ import annotations + +from typing import cast + +from homeassistant.const import CONF_PLATFORM +from homeassistant.core import CALLBACK_TYPE, HomeAssistant +from homeassistant.helpers.trigger import ( + TriggerActionType, + TriggerInfo, + TriggerProtocol, +) +from homeassistant.helpers.typing import ConfigType + +from .triggers import turn_on + +TRIGGERS = { + "turn_on": turn_on, +} + + +def _get_trigger_platform(config: ConfigType) -> TriggerProtocol: + """Return trigger platform.""" + platform_split = config[CONF_PLATFORM].split(".", maxsplit=1) + if len(platform_split) < 2 or platform_split[1] not in TRIGGERS: + raise ValueError( + f"Unknown LG Netcast TV trigger platform {config[CONF_PLATFORM]}" + ) + return cast(TriggerProtocol, TRIGGERS[platform_split[1]]) + + +async def async_validate_trigger_config( + hass: HomeAssistant, config: ConfigType +) -> ConfigType: + """Validate config.""" + platform = _get_trigger_platform(config) + return cast(ConfigType, platform.TRIGGER_SCHEMA(config)) + + +async def async_attach_trigger( + hass: HomeAssistant, + config: ConfigType, + action: TriggerActionType, + trigger_info: TriggerInfo, +) -> CALLBACK_TYPE: + """Attach trigger of specified platform.""" + platform = _get_trigger_platform(config) + return await platform.async_attach_trigger(hass, config, action, trigger_info) diff --git a/homeassistant/components/lg_netcast/triggers/__init__.py b/homeassistant/components/lg_netcast/triggers/__init__.py new file mode 100644 index 00000000000000..d352620118ed2f --- /dev/null +++ b/homeassistant/components/lg_netcast/triggers/__init__.py @@ -0,0 +1 @@ +"""LG Netcast triggers.""" diff --git a/homeassistant/components/lg_netcast/triggers/turn_on.py b/homeassistant/components/lg_netcast/triggers/turn_on.py new file mode 100644 index 00000000000000..118ed89797e58b --- /dev/null +++ b/homeassistant/components/lg_netcast/triggers/turn_on.py @@ -0,0 +1,115 @@ +"""LG Netcast TV device turn on trigger.""" + +import voluptuous as vol + +from homeassistant.const import ( + ATTR_DEVICE_ID, + ATTR_ENTITY_ID, + CONF_DEVICE_ID, + CONF_DOMAIN, + CONF_PLATFORM, + CONF_TYPE, +) +from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback +from homeassistant.helpers import config_validation as cv, entity_registry as er +from homeassistant.helpers.trigger import ( + PluggableAction, + TriggerActionType, + TriggerInfo, +) +from homeassistant.helpers.typing import ConfigType + +from ..const import DOMAIN +from ..helpers import async_get_device_entry_by_device_id + +PLATFORM_TYPE = f"{DOMAIN}.{__name__.rsplit('.', maxsplit=1)[-1]}" + +TRIGGER_SCHEMA = vol.All( + cv.TRIGGER_BASE_SCHEMA.extend( + { + vol.Required(CONF_PLATFORM): PLATFORM_TYPE, + vol.Optional(ATTR_DEVICE_ID): vol.All(cv.ensure_list, [cv.string]), + vol.Optional(ATTR_ENTITY_ID): cv.entity_ids, + }, + ), + cv.has_at_least_one_key(ATTR_ENTITY_ID, ATTR_DEVICE_ID), +) + + +def async_get_turn_on_trigger(device_id: str) -> dict[str, str]: + """Return data for a turn on trigger.""" + + return { + CONF_PLATFORM: "device", + CONF_DEVICE_ID: device_id, + CONF_DOMAIN: DOMAIN, + CONF_TYPE: PLATFORM_TYPE, + } + + +async def async_attach_trigger( + hass: HomeAssistant, + config: ConfigType, + action: TriggerActionType, + trigger_info: TriggerInfo, + *, + platform_type: str = PLATFORM_TYPE, +) -> CALLBACK_TYPE | None: + """Attach a trigger.""" + device_ids = set() + if ATTR_DEVICE_ID in config: + device_ids.update(config.get(ATTR_DEVICE_ID, [])) + + if ATTR_ENTITY_ID in config: + ent_reg = er.async_get(hass) + + def _get_device_id_from_entity_id(entity_id): + entity_entry = ent_reg.async_get(entity_id) + + if ( + entity_entry is None + or entity_entry.device_id is None + or entity_entry.platform != DOMAIN + ): + raise ValueError(f"Entity {entity_id} is not a valid {DOMAIN} entity.") + + return entity_entry.device_id + + device_ids.update( + { + _get_device_id_from_entity_id(entity_id) + for entity_id in config.get(ATTR_ENTITY_ID, []) + } + ) + + trigger_data = trigger_info["trigger_data"] + + unsubs = [] + + for device_id in device_ids: + device = async_get_device_entry_by_device_id(hass, device_id) + device_name = device.name_by_user or device.name + + variables = { + **trigger_data, + CONF_PLATFORM: platform_type, + ATTR_DEVICE_ID: device_id, + "description": f"lg netcast turn on trigger for {device_name}", + } + + turn_on_trigger = async_get_turn_on_trigger(device_id) + + unsubs.append( + PluggableAction.async_attach_trigger( + hass, turn_on_trigger, action, {"trigger": variables} + ) + ) + + @callback + def async_remove() -> None: + """Remove state listeners async.""" + for unsub in unsubs: + unsub() + unsubs.clear() + + return async_remove diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 125f02df3b50fc..d1fe540c1b4af0 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -285,6 +285,7 @@ "ld2410_ble", "leaone", "led_ble", + "lg_netcast", "lg_soundbar", "lidarr", "lifx", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index 340be50978dc50..1b964ceae34f54 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -3177,8 +3177,8 @@ "name": "LG", "integrations": { "lg_netcast": { - "integration_type": "hub", - "config_flow": false, + "integration_type": "device", + "config_flow": true, "iot_class": "local_polling", "name": "LG Netcast" }, diff --git a/requirements_all.txt b/requirements_all.txt index 653e481d2fb08f..92c2533dc4d878 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -1931,7 +1931,7 @@ pylast==5.1.0 pylaunches==1.4.0 # homeassistant.components.lg_netcast -pylgnetcast==0.3.7 +pylgnetcast==0.3.9 # homeassistant.components.forked_daapd pylibrespot-java==0.1.1 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 0decf82fe0cb7e..216edd0c5da45a 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1502,6 +1502,9 @@ pylast==5.1.0 # homeassistant.components.launch_library pylaunches==1.4.0 +# homeassistant.components.lg_netcast +pylgnetcast==0.3.9 + # homeassistant.components.forked_daapd pylibrespot-java==0.1.1 diff --git a/tests/components/lg_netcast/__init__.py b/tests/components/lg_netcast/__init__.py new file mode 100644 index 00000000000000..ce3e09aeb6556a --- /dev/null +++ b/tests/components/lg_netcast/__init__.py @@ -0,0 +1,116 @@ +"""Tests for LG Netcast TV.""" + +from unittest.mock import patch +from xml.etree import ElementTree + +from pylgnetcast import AccessTokenError, LgNetCastClient, SessionIdError +import requests + +from homeassistant.components.lg_netcast import DOMAIN +from homeassistant.components.media_player import DOMAIN as MP_DOMAIN +from homeassistant.const import ( + CONF_ACCESS_TOKEN, + CONF_HOST, + CONF_ID, + CONF_MODEL, + CONF_NAME, +) +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry + +FAIL_TO_BIND_IP = "1.2.3.4" + +IP_ADDRESS = "192.168.1.239" +DEVICE_TYPE = "TV" +MODEL_NAME = "MockLGModelName" +FRIENDLY_NAME = "LG Smart TV" +UNIQUE_ID = "1234" +ENTITY_ID = f"{MP_DOMAIN}.{MODEL_NAME.lower()}" + +FAKE_SESSION_ID = "987654321" +FAKE_PIN = "123456" + + +def _patched_lgnetcast_client( + *args, + session_error=False, + fail_connection: bool = True, + invalid_details: bool = False, + always_404: bool = False, + no_unique_id: bool = False, + **kwargs, +): + client = LgNetCastClient(*args, **kwargs) + + def _get_fake_session_id(): + if not client.access_token: + raise AccessTokenError("Fake Access Token Requested") + if session_error: + raise SessionIdError("Can not get session id from TV.") + return FAKE_SESSION_ID + + def _get_fake_query_device_info(): + if fail_connection: + raise requests.exceptions.ConnectTimeout("Mocked Failed Connection") + if always_404: + return None + if invalid_details: + raise ElementTree.ParseError("Mocked Parsed Error") + return { + "uuid": UNIQUE_ID if not no_unique_id else None, + "model_name": MODEL_NAME, + "friendly_name": FRIENDLY_NAME, + } + + client._get_session_id = _get_fake_session_id + client.query_device_info = _get_fake_query_device_info + + return client + + +def _patch_lg_netcast( + *, + session_error: bool = False, + fail_connection: bool = False, + invalid_details: bool = False, + always_404: bool = False, + no_unique_id: bool = False, +): + def _generate_fake_lgnetcast_client(*args, **kwargs): + return _patched_lgnetcast_client( + *args, + session_error=session_error, + fail_connection=fail_connection, + invalid_details=invalid_details, + always_404=always_404, + no_unique_id=no_unique_id, + **kwargs, + ) + + return patch( + "homeassistant.components.lg_netcast.config_flow.LgNetCastClient", + new=_generate_fake_lgnetcast_client, + ) + + +async def setup_lgnetcast(hass: HomeAssistant, unique_id: str = UNIQUE_ID): + """Initialize lg netcast and media_player for tests.""" + config_entry = MockConfigEntry( + domain=DOMAIN, + data={ + CONF_HOST: IP_ADDRESS, + CONF_ACCESS_TOKEN: FAKE_PIN, + CONF_NAME: MODEL_NAME, + CONF_MODEL: MODEL_NAME, + CONF_ID: unique_id, + }, + title=MODEL_NAME, + unique_id=unique_id, + ) + config_entry.add_to_hass(hass) + + await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + return config_entry diff --git a/tests/components/lg_netcast/conftest.py b/tests/components/lg_netcast/conftest.py new file mode 100644 index 00000000000000..4faee2c6f06289 --- /dev/null +++ b/tests/components/lg_netcast/conftest.py @@ -0,0 +1,11 @@ +"""Common fixtures and objects for the LG Netcast integration tests.""" + +import pytest + +from tests.common import async_mock_service + + +@pytest.fixture +def calls(hass): + """Track calls to a mock service.""" + return async_mock_service(hass, "test", "automation") diff --git a/tests/components/lg_netcast/test_config_flow.py b/tests/components/lg_netcast/test_config_flow.py new file mode 100644 index 00000000000000..c159b8fb9d2c9a --- /dev/null +++ b/tests/components/lg_netcast/test_config_flow.py @@ -0,0 +1,252 @@ +"""Define tests for the LG Netcast config flow.""" + +from datetime import timedelta +from unittest.mock import DEFAULT, patch + +from homeassistant import data_entry_flow +from homeassistant.components.lg_netcast.const import DOMAIN +from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER +from homeassistant.const import ( + CONF_ACCESS_TOKEN, + CONF_HOST, + CONF_ID, + CONF_MODEL, + CONF_NAME, +) +from homeassistant.core import HomeAssistant + +from . import ( + FAKE_PIN, + FRIENDLY_NAME, + IP_ADDRESS, + MODEL_NAME, + UNIQUE_ID, + _patch_lg_netcast, +) + +from tests.common import MockConfigEntry + + +async def test_show_form(hass: HomeAssistant) -> None: + """Test that the form is served with no input.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + assert result["type"] == data_entry_flow.FlowResultType.FORM + assert result["step_id"] == "user" + + +async def test_user_invalid_host(hass: HomeAssistant) -> None: + """Test that errors are shown when the host is invalid.""" + with _patch_lg_netcast(): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "invalid/host"} + ) + + assert result["errors"] == {CONF_HOST: "invalid_host"} + + +async def test_manual_host(hass: HomeAssistant) -> None: + """Test manual host configuration.""" + with _patch_lg_netcast(): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: IP_ADDRESS} + ) + + assert result["type"] == data_entry_flow.FlowResultType.FORM + assert result["step_id"] == "authorize" + assert not result["errors"] + + result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {}) + assert result2["type"] == data_entry_flow.FlowResultType.FORM + assert result2["step_id"] == "authorize" + assert result2["errors"] is not None + assert result2["errors"][CONF_ACCESS_TOKEN] == "invalid_access_token" + + result3 = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_ACCESS_TOKEN: FAKE_PIN} + ) + + assert result3["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY + assert result3["title"] == FRIENDLY_NAME + assert result3["data"] == { + CONF_HOST: IP_ADDRESS, + CONF_ACCESS_TOKEN: FAKE_PIN, + CONF_NAME: FRIENDLY_NAME, + CONF_MODEL: MODEL_NAME, + CONF_ID: UNIQUE_ID, + } + + +async def test_manual_host_no_connection_during_authorize(hass: HomeAssistant) -> None: + """Test manual host configuration.""" + with _patch_lg_netcast(fail_connection=True): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: IP_ADDRESS} + ) + + assert result["type"] == data_entry_flow.FlowResultType.ABORT + assert result["reason"] == "cannot_connect" + + +async def test_manual_host_invalid_details_during_authorize( + hass: HomeAssistant, +) -> None: + """Test manual host configuration.""" + with _patch_lg_netcast(invalid_details=True): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: IP_ADDRESS} + ) + + assert result["type"] == data_entry_flow.FlowResultType.ABORT + assert result["reason"] == "cannot_connect" + + +async def test_manual_host_unsuccessful_details_response(hass: HomeAssistant) -> None: + """Test manual host configuration.""" + with _patch_lg_netcast(always_404=True): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: IP_ADDRESS} + ) + + assert result["type"] == data_entry_flow.FlowResultType.ABORT + assert result["reason"] == "cannot_connect" + + +async def test_manual_host_no_unique_id_response(hass: HomeAssistant) -> None: + """Test manual host configuration.""" + with _patch_lg_netcast(no_unique_id=True): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: IP_ADDRESS} + ) + + assert result["type"] == data_entry_flow.FlowResultType.ABORT + assert result["reason"] == "invalid_host" + + +async def test_invalid_session_id(hass: HomeAssistant) -> None: + """Test Invalid Session ID.""" + with _patch_lg_netcast(session_error=True): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: IP_ADDRESS} + ) + + assert result["type"] == data_entry_flow.FlowResultType.FORM + assert result["step_id"] == "authorize" + assert not result["errors"] + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_ACCESS_TOKEN: FAKE_PIN} + ) + + assert result2["type"] == data_entry_flow.FlowResultType.FORM + assert result2["step_id"] == "authorize" + assert result2["errors"] is not None + assert result2["errors"]["base"] == "cannot_connect" + + +async def test_import(hass: HomeAssistant) -> None: + """Test that the import works.""" + with _patch_lg_netcast(): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_IMPORT}, + data={ + CONF_HOST: IP_ADDRESS, + CONF_ACCESS_TOKEN: FAKE_PIN, + CONF_NAME: MODEL_NAME, + }, + ) + + assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY + assert result["result"].unique_id == UNIQUE_ID + assert result["data"] == { + CONF_HOST: IP_ADDRESS, + CONF_ACCESS_TOKEN: FAKE_PIN, + CONF_NAME: MODEL_NAME, + CONF_MODEL: MODEL_NAME, + CONF_ID: UNIQUE_ID, + } + + +async def test_import_not_online(hass: HomeAssistant) -> None: + """Test that the import works.""" + with _patch_lg_netcast(fail_connection=True): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_IMPORT}, + data={ + CONF_HOST: IP_ADDRESS, + CONF_ACCESS_TOKEN: FAKE_PIN, + CONF_NAME: MODEL_NAME, + }, + ) + + assert result["type"] == data_entry_flow.FlowResultType.ABORT + assert result["reason"] == "cannot_connect" + + +async def test_import_duplicate_error(hass): + """Test that errors are shown when duplicates are added during import.""" + config_entry = MockConfigEntry( + domain=DOMAIN, + unique_id=UNIQUE_ID, + data={ + CONF_HOST: IP_ADDRESS, + CONF_ACCESS_TOKEN: FAKE_PIN, + CONF_NAME: MODEL_NAME, + CONF_ID: UNIQUE_ID, + }, + ) + config_entry.add_to_hass(hass) + + with _patch_lg_netcast(): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_IMPORT}, + data={ + CONF_HOST: IP_ADDRESS, + CONF_ACCESS_TOKEN: FAKE_PIN, + CONF_NAME: MODEL_NAME, + CONF_ID: UNIQUE_ID, + }, + ) + + assert result["type"] == data_entry_flow.FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_display_access_token_aborted(hass: HomeAssistant): + """Test Access token display is cancelled.""" + + def _async_track_time_interval( + hass: HomeAssistant, + action, + interval: timedelta, + *, + name=None, + cancel_on_shutdown=None, + ): + hass.async_create_task(action()) + return DEFAULT + + with ( + _patch_lg_netcast(session_error=True), + patch( + "homeassistant.components.lg_netcast.config_flow.async_track_time_interval" + ) as mock_interval, + ): + mock_interval.side_effect = _async_track_time_interval + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: IP_ADDRESS} + ) + + assert result["type"] == data_entry_flow.FlowResultType.FORM + assert result["step_id"] == "authorize" + assert not result["errors"] + + assert mock_interval.called + + hass.config_entries.flow.async_abort(result["flow_id"]) + assert mock_interval.return_value.called diff --git a/tests/components/lg_netcast/test_device_trigger.py b/tests/components/lg_netcast/test_device_trigger.py new file mode 100644 index 00000000000000..05911acc41dafe --- /dev/null +++ b/tests/components/lg_netcast/test_device_trigger.py @@ -0,0 +1,148 @@ +"""The tests for LG NEtcast device triggers.""" + +import pytest + +from homeassistant.components import automation +from homeassistant.components.device_automation import DeviceAutomationType +from homeassistant.components.device_automation.exceptions import ( + InvalidDeviceAutomationConfig, +) +from homeassistant.components.lg_netcast import DOMAIN, device_trigger +from homeassistant.config_entries import ConfigEntryState +from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import device_registry as dr +from homeassistant.setup import async_setup_component + +from . import ENTITY_ID, UNIQUE_ID, setup_lgnetcast + +from tests.common import MockConfigEntry, async_get_device_automations + + +async def test_get_triggers( + hass: HomeAssistant, device_registry: dr.DeviceRegistry +) -> None: + """Test we get the expected triggers.""" + await setup_lgnetcast(hass) + + device = device_registry.async_get_device(identifiers={(DOMAIN, UNIQUE_ID)}) + assert device is not None + + turn_on_trigger = { + "platform": "device", + "domain": DOMAIN, + "type": "lg_netcast.turn_on", + "device_id": device.id, + "metadata": {}, + } + + triggers = await async_get_device_automations( + hass, DeviceAutomationType.TRIGGER, device.id + ) + assert turn_on_trigger in triggers + + +async def test_if_fires_on_turn_on_request( + hass: HomeAssistant, calls: list[ServiceCall], device_registry: dr.DeviceRegistry +) -> None: + """Test for turn_on triggers firing.""" + await setup_lgnetcast(hass) + + device = device_registry.async_get_device(identifiers={(DOMAIN, UNIQUE_ID)}) + assert device is not None + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + "platform": "device", + "domain": DOMAIN, + "device_id": device.id, + "type": "lg_netcast.turn_on", + }, + "action": { + "service": "test.automation", + "data_template": { + "some": "{{ trigger.device_id }}", + "id": "{{ trigger.id }}", + }, + }, + }, + { + "trigger": { + "platform": "lg_netcast.turn_on", + "entity_id": ENTITY_ID, + }, + "action": { + "service": "test.automation", + "data_template": { + "some": ENTITY_ID, + "id": "{{ trigger.id }}", + }, + }, + }, + ], + }, + ) + + await hass.services.async_call( + "media_player", + "turn_on", + {"entity_id": ENTITY_ID}, + blocking=True, + ) + + await hass.async_block_till_done() + assert len(calls) == 2 + assert calls[0].data["some"] == device.id + assert calls[0].data["id"] == 0 + assert calls[1].data["some"] == ENTITY_ID + assert calls[1].data["id"] == 0 + + +async def test_failure_scenarios( + hass: HomeAssistant, device_registry: dr.DeviceRegistry +) -> None: + """Test failure scenarios.""" + await setup_lgnetcast(hass) + + # Test wrong trigger platform type + with pytest.raises(HomeAssistantError): + await device_trigger.async_attach_trigger( + hass, {"type": "wrong.type", "device_id": "invalid_device_id"}, None, {} + ) + + # Test invalid device id + with pytest.raises(HomeAssistantError): + await device_trigger.async_validate_trigger_config( + hass, + { + "platform": "device", + "domain": DOMAIN, + "type": "lg_netcast.turn_on", + "device_id": "invalid_device_id", + }, + ) + + entry = MockConfigEntry(domain="fake", state=ConfigEntryState.LOADED, data={}) + entry.add_to_hass(hass) + + device = device_registry.async_get_or_create( + config_entry_id=entry.entry_id, identifiers={("fake", "fake")} + ) + + config = { + "platform": "device", + "domain": DOMAIN, + "device_id": device.id, + "type": "lg_netcast.turn_on", + } + + # Test that device id from non lg_netcast domain raises exception + with pytest.raises(InvalidDeviceAutomationConfig): + await device_trigger.async_validate_trigger_config(hass, config) + + # Test that only valid triggers are attached diff --git a/tests/components/lg_netcast/test_trigger.py b/tests/components/lg_netcast/test_trigger.py new file mode 100644 index 00000000000000..e75dac501c3d18 --- /dev/null +++ b/tests/components/lg_netcast/test_trigger.py @@ -0,0 +1,189 @@ +"""The tests for LG Netcast device triggers.""" + +from unittest.mock import patch + +import pytest + +from homeassistant.components import automation +from homeassistant.components.lg_netcast import DOMAIN +from homeassistant.const import SERVICE_RELOAD +from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import device_registry as dr +from homeassistant.setup import async_setup_component + +from . import ENTITY_ID, UNIQUE_ID, setup_lgnetcast + +from tests.common import MockEntity, MockEntityPlatform + + +async def test_lg_netcast_turn_on_trigger_device_id( + hass: HomeAssistant, calls: list[ServiceCall], device_registry: dr.DeviceRegistry +) -> None: + """Test for turn_on trigger by device_id firing.""" + await setup_lgnetcast(hass) + + device = device_registry.async_get_device(identifiers={(DOMAIN, UNIQUE_ID)}) + assert device, repr(device_registry.devices) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + "platform": "lg_netcast.turn_on", + "device_id": device.id, + }, + "action": { + "service": "test.automation", + "data_template": { + "some": device.id, + "id": "{{ trigger.id }}", + }, + }, + }, + ], + }, + ) + + await hass.services.async_call( + "media_player", + "turn_on", + {"entity_id": ENTITY_ID}, + blocking=True, + ) + await hass.async_block_till_done() + + assert len(calls) == 1 + assert calls[0].data["some"] == device.id + assert calls[0].data["id"] == 0 + + with patch("homeassistant.config.load_yaml_dict", return_value={}): + await hass.services.async_call(automation.DOMAIN, SERVICE_RELOAD, blocking=True) + + calls.clear() + + with pytest.raises(HomeAssistantError): + await hass.services.async_call( + "media_player", + "turn_on", + {"entity_id": ENTITY_ID}, + blocking=True, + ) + + await hass.async_block_till_done() + assert len(calls) == 0 + + +async def test_lg_netcast_turn_on_trigger_entity_id(hass: HomeAssistant, calls): + """Test for turn_on triggers by entity firing.""" + await setup_lgnetcast(hass) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + "platform": "lg_netcast.turn_on", + "entity_id": ENTITY_ID, + }, + "action": { + "service": "test.automation", + "data_template": { + "some": ENTITY_ID, + "id": "{{ trigger.id }}", + }, + }, + }, + ], + }, + ) + + await hass.services.async_call( + "media_player", + "turn_on", + {"entity_id": ENTITY_ID}, + blocking=True, + ) + await hass.async_block_till_done() + + assert len(calls) == 1 + assert calls[0].data["some"] == ENTITY_ID + assert calls[0].data["id"] == 0 + + +async def test_wrong_trigger_platform_type( + hass: HomeAssistant, caplog: pytest.LogCaptureFixture +) -> None: + """Test wrong trigger platform type.""" + await setup_lgnetcast(hass) + + await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + "platform": "lg_netcast.wrong_type", + "entity_id": ENTITY_ID, + }, + "action": { + "service": "test.automation", + "data_template": { + "some": ENTITY_ID, + "id": "{{ trigger.id }}", + }, + }, + }, + ], + }, + ) + + assert ( + "ValueError: Unknown LG Netcast TV trigger platform lg_netcast.wrong_type" + in caplog.text + ) + + +async def test_trigger_invalid_entity_id( + hass: HomeAssistant, caplog: pytest.LogCaptureFixture +) -> None: + """Test turn on trigger using invalid entity_id.""" + await setup_lgnetcast(hass) + + platform = MockEntityPlatform(hass) + + invalid_entity = f"{DOMAIN}.invalid" + await platform.async_add_entities([MockEntity(name=invalid_entity)]) + + await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + "platform": "lg_netcast.turn_on", + "entity_id": invalid_entity, + }, + "action": { + "service": "test.automation", + "data_template": { + "some": ENTITY_ID, + "id": "{{ trigger.id }}", + }, + }, + } + ], + }, + ) + + assert ( + f"ValueError: Entity {invalid_entity} is not a valid lg_netcast entity" + in caplog.text + )