evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Config Flow to LG Netcast (#104913)

Browse source 
* 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 <erik@montnemery.com>
This commit is contained in:
Stephen Alderman 2024-04-16 08:29:02 +01:00 • committed by GitHub
parent a99ecb024e
commit f62fb76765
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 1411 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

3
CODEOWNERS
View file

@ -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

32
homeassistant/components/lg_netcast/__init__.py
View file

@ -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

217
homeassistant/components/lg_netcast/config_flow.py Normal file
View file

@ -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
)

6
homeassistant/components/lg_netcast/const.py
View file

@ -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"

88
homeassistant/components/lg_netcast/device_trigger.py Normal file
View file

@ -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}")

59
homeassistant/components/lg_netcast/helpers.py Normal file
View file

@ -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

7
homeassistant/components/lg_netcast/manifest.json
View file

@ -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"]
}

92
homeassistant/components/lg_netcast/media_player.py
View file

@ -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 @@ from homeassistant.components.media_player import (
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 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
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 @@ class LgTVDevice(MediaPlayerEntity):
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 @@ class LgTVDevice(MediaPlayerEntity):
@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 @@ class LgTVDevice(MediaPlayerEntity):
"""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."""

46
homeassistant/components/lg_netcast/strings.json Normal file
View file

@ -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"
}
}
}

49
homeassistant/components/lg_netcast/trigger.py Normal file
View file

@ -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)

1
homeassistant/components/lg_netcast/triggers/__init__.py Normal file
View file

@ -0,0 +1 @@
"""LG Netcast triggers."""

115
homeassistant/components/lg_netcast/triggers/turn_on.py Normal file
View file

@ -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

1
homeassistant/generated/config_flows.py
View file

@ -285,6 +285,7 @@ FLOWS = {
"ld2410_ble",
"leaone",
"led_ble",
"lg_netcast",
"lg_soundbar",
"lidarr",
"lifx",

4
homeassistant/generated/integrations.json
View file

@ -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"
},

2
requirements_all.txt
View file

@ -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

3
requirements_test_all.txt
View file

@ -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

116
tests/components/lg_netcast/__init__.py Normal file
View file

@ -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

11
tests/components/lg_netcast/conftest.py Normal file
View file

@ -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")

252
tests/components/lg_netcast/test_config_flow.py Normal file
View file

@ -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

148
tests/components/lg_netcast/test_device_trigger.py Normal file
View file

@ -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

189
tests/components/lg_netcast/test_trigger.py Normal file
View file

@ -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
)