starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-10-06 14:29:21 -04:00
Code Issues Packages Projects Releases Wiki Activity

Migrate YouTube channels to config subentries (#181438)

Browse Source
This commit is contained in:
DevHugo
2026-10-01 20:27:40 +01:00
committed by GitHub
parent 8f1570c8b7
commit c4b0a4e1ce
12 changed files with 1205 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/youtube/__init__.py
+93 -15
View File
@@ -1,14 +1,18 @@
"""Support for YouTube."""
from types import MappingProxyType
from homeassistant.config_entries import ConfigEntry, ConfigSubentry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.config_entry_oauth2_flow import (
OAuth2Session,
async_get_config_entry_implementation,
)
from .api import AsyncConfigEntryAuth
from .const import CONF_CHANNEL_ID, CONF_CHANNELS, DOMAIN, SUBENTRY_TYPE_CHANNEL
from .coordinator import YouTubeConfigEntry, YouTubeDataUpdateCoordinator
PLATFORMS = [Platform.SENSOR]
@@ -20,13 +24,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: YouTubeConfigEntry) -> b
session = OAuth2Session(hass, entry, implementation)
auth = AsyncConfigEntryAuth(hass, session)
await auth.check_and_refresh_token()
coordinator = YouTubeDataUpdateCoordinator(hass, entry, auth)
coordinator = YouTubeDataUpdateCoordinator(hass, entry, auth)
await coordinator.async_config_entry_first_refresh()
await delete_devices(hass, entry, coordinator)
entry.runtime_data = coordinator
entry.async_on_unload(entry.add_update_listener(async_update_listener))
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
@@ -37,15 +40,90 @@ async def async_unload_entry(hass: HomeAssistant, entry: YouTubeConfigEntry) ->
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def delete_devices(
hass: HomeAssistant,
entry: YouTubeConfigEntry,
coordinator: YouTubeDataUpdateCoordinator,
) -> None:
"""Delete all devices created by integration."""
channel_ids = list(coordinator.data)
async def async_update_listener(hass: HomeAssistant, entry: YouTubeConfigEntry) -> None:
"""Reload the config entry when channels are added or removed."""
subentry_ids = {
subentry.subentry_id
for subentry in entry.get_subentries_of_type(SUBENTRY_TYPE_CHANNEL)
}
if subentry_ids == entry.runtime_data.subentry_ids:
# Token refreshes update the entry data only; the coordinator
# uses the refreshed token without needing a reload.
return
await hass.config_entries.async_reload(entry.entry_id)
async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Migrate old entries to the subentry structure."""
device_registry = dr.async_get(hass)
dev_entries = dr.async_entries_for_config_entry(device_registry, entry.entry_id)
for dev_entry in dev_entries:
if any(identifier[1] in channel_ids for identifier in dev_entry.identifiers):
device_registry.async_remove_device(dev_entry.id)
entity_registry = er.async_get(hass)
prefix = f"{entry.entry_id}_"
channel_ids = dict.fromkeys(entry.options.get(CONF_CHANNELS, []))
subentries: dict[str, ConfigSubentry] = {}
for channel_id in channel_ids:
device = device_registry.async_get_device_by_identifier(
(DOMAIN, f"{prefix}{channel_id}"), entry.entry_id
)
title = channel_id
if device is not None and device.name is not None:
# Prefer the channel name of the existing device so the title
# survives even if the channel can no longer be fetched.
title = device.name
subentry = ConfigSubentry(
data=MappingProxyType({CONF_CHANNEL_ID: channel_id}),
subentry_type=SUBENTRY_TYPE_CHANNEL,
title=title,
unique_id=channel_id,
)
hass.config_entries.async_add_subentry(entry, subentry)
subentries[channel_id] = subentry
# Attach the entities of tracked channels to their subentry and remove
# entities left behind by channels which are no longer tracked.
channel_prefixes = {
f"{prefix}{channel_id}_": channel_id for channel_id in channel_ids
}
for entity_entry in er.async_entries_for_config_entry(
entity_registry, entry.entry_id
):
channel_subentry = next(
(
subentries[channel_id]
for channel_prefix, channel_id in channel_prefixes.items()
if entity_entry.unique_id.startswith(channel_prefix)
),
None,
)
if channel_subentry is None:
entity_registry.async_remove(entity_entry.entity_id)
else:
entity_registry.async_update_entity(
entity_entry.entity_id,
config_subentry_id=channel_subentry.subentry_id,
)
# Move the devices of tracked channels to their subentry and remove
# devices left behind by channels which are no longer tracked.
for device_entry in dr.async_entries_for_config_entry(
device_registry, entry.entry_id
):
channel_id = next(
(
identifier[1].removeprefix(prefix)
for identifier in device_entry.identifiers
if identifier[0] == DOMAIN and identifier[1].startswith(prefix)
),
None,
)
if channel_id is not None and channel_id in subentries:
device_registry.async_update_device(
device_entry.id,
new_identifiers={(DOMAIN, channel_id)},
new_config_subentry_id=subentries[channel_id].subentry_id,
)
else:
device_registry.async_remove_device(device_entry.id)
hass.config_entries.async_update_entry(entry, version=2, options={})
return True
homeassistant/components/youtube/config_flow.py
+182 -97
View File
@@ -8,47 +8,105 @@ import probatio
from youtubeaio.types import AuthScope, ForbiddenError
from youtubeaio.youtube import YouTube
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult, OptionsFlow
from homeassistant.config_entries import (
SOURCE_REAUTH,
ConfigEntry,
ConfigFlowResult,
ConfigSubentryData,
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_TOKEN
from homeassistant.core import callback
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.config_entry_oauth2_flow import (
OAuth2Session,
async_get_config_entry_implementation,
)
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .api import AsyncConfigEntryAuth
from .const import (
CHANNEL_CREATION_HELP_URL,
CONF_CHANNEL_ID,
CONF_CHANNELS,
DEFAULT_ACCESS,
DOMAIN,
LOGGER,
SUBENTRY_TYPE_CHANNEL,
)
from .coordinator import YouTubeConfigEntry
async def async_get_channel_options(
youtube: YouTube,
) -> tuple[list[SelectOptionDict], dict[str, str], bool]:
"""List the channels the user can track.
Returns the selectable options, a mapping of channel id to title, and
whether the user has their own channel.
"""
own_channels = [
channel
async for channel in youtube.get_user_channels()
if channel.snippet is not None
]
subscriptions = [
subscription
async for subscription in youtube.get_user_subscriptions()
if subscription.snippet is not None
]
selectable_channels = [
SelectOptionDict(
value=channel.channel_id,
label=f"{channel.snippet.title} (Your Channel)",
)
for channel in own_channels
]
selectable_channels.extend(
SelectOptionDict(
value=subscription.snippet.channel_id,
label=subscription.snippet.title,
)
for subscription in subscriptions
)
channel_titles = {
subscription.snippet.channel_id: subscription.snippet.title
for subscription in subscriptions
} | {channel.channel_id: channel.snippet.title for channel in own_channels}
return selectable_channels, channel_titles, bool(own_channels)
class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Config flow to handle Google OAuth2 authentication."""
VERSION = 2
_data: dict[str, Any] = {}
_title: str = ""
_channel_titles: dict[str, str] = {}
DOMAIN = DOMAIN
_youtube: YouTube | None = None
@staticmethod
@classmethod
@callback
@override
def async_get_options_flow(
config_entry: YouTubeConfigEntry,
) -> YouTubeOptionsFlowHandler:
"""Get the options flow for this handler."""
return YouTubeOptionsFlowHandler()
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this handler."""
return {SUBENTRY_TYPE_CHANNEL: ChannelFlowHandler}
@property
@override
@@ -133,120 +191,147 @@ class OAuth2FlowHandler(
) -> ConfigFlowResult:
"""Select which channels to track."""
if user_input:
channel_ids = dict.fromkeys(user_input[CONF_CHANNELS])
return self.async_create_entry(
title=self._title,
data=self._data,
options=user_input,
subentries=[
ConfigSubentryData(
data={CONF_CHANNEL_ID: channel_id},
subentry_type=SUBENTRY_TYPE_CHANNEL,
title=self._channel_titles[channel_id],
unique_id=channel_id,
)
for channel_id in channel_ids
],
)
youtube = await self.get_resource(self._data[CONF_TOKEN][CONF_ACCESS_TOKEN])
# Get user's own channels
own_channels = [
channel
async for channel in youtube.get_user_channels()
if channel.snippet is not None
]
if not own_channels:
try:
youtube = YouTube(session=async_get_clientsession(self.hass))
await youtube.set_user_authentication(
self._data[CONF_TOKEN][CONF_ACCESS_TOKEN], [AuthScope.READ_ONLY]
)
(
selectable_channels,
channel_titles,
_has_own_channel,
) = await async_get_channel_options(youtube)
except ForbiddenError as ex:
error = ex.args[0]
return self.async_abort(
reason="no_channel",
description_placeholders={"support_url": CHANNEL_CREATION_HELP_URL},
reason="access_not_configured",
description_placeholders={"message": error},
)
# Start with user's own channels
selectable_channels = [
SelectOptionDict(
value=channel.channel_id,
label=f"{channel.snippet.title} (Your Channel)",
)
for channel in own_channels
]
# Add subscribed channels
selectable_channels.extend(
[
SelectOptionDict(
value=subscription.snippet.channel_id,
label=subscription.snippet.title,
)
async for subscription in youtube.get_user_subscriptions()
]
)
if not selectable_channels:
return self.async_abort(reason="no_subscriptions")
except Exception as ex: # noqa: BLE001
LOGGER.error("Unknown error occurred: %s", ex.args)
return self.async_abort(reason="unknown")
self._channel_titles = channel_titles
return self.async_show_form(
step_id="channels",
data_schema=probatio.Schema(
{
probatio.Required(CONF_CHANNELS): SelectSelector(
SelectSelectorConfig(options=selectable_channels, multiple=True)
SelectSelectorConfig(
options=selectable_channels,
multiple=True,
mode=SelectSelectorMode.DROPDOWN,
)
),
}
),
)
class YouTubeOptionsFlowHandler(OptionsFlow):
"""YouTube Options flow handler."""
class ChannelFlowHandler(ConfigSubentryFlow):
"""Handle subentry flow for adding a channel."""
async def async_step_init(
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Initialize form."""
) -> SubentryFlowResult:
"""User flow to add a channel."""
if user_input is not None:
return self.async_create_entry(
title=self.config_entry.title,
data=user_input,
return await self._async_create_entry(user_input[CONF_CHANNEL_ID])
config_entry: YouTubeConfigEntry = self._get_entry()
try:
(
selectable_channels,
_channel_titles,
_has_own_channel,
) = await async_get_channel_options(
await self._async_get_youtube(config_entry)
)
youtube = YouTube(session=async_get_clientsession(self.hass))
await youtube.set_user_authentication(
self.config_entry.data[CONF_TOKEN][CONF_ACCESS_TOKEN], [AuthScope.READ_ONLY]
)
# Get user's own channels
own_channels = [
channel
async for channel in youtube.get_user_channels()
if channel.snippet is not None
]
if not own_channels:
except ForbiddenError as ex:
error = ex.args[0]
return self.async_abort(
reason="no_channel",
description_placeholders={"support_url": CHANNEL_CREATION_HELP_URL},
reason="access_not_configured",
description_placeholders={"message": error},
)
except Exception as ex: # noqa: BLE001
LOGGER.error("Unknown error occurred: %s", ex.args)
return self.async_abort(reason="unknown")
# Start with user's own channels
selectable_channels = [
SelectOptionDict(
value=channel.channel_id,
label=f"{channel.snippet.title} (Your Channel)",
)
for channel in own_channels
]
# Add subscribed channels
selectable_channels.extend(
[
SelectOptionDict(
value=subscription.snippet.channel_id,
label=subscription.snippet.title,
)
async for subscription in youtube.get_user_subscriptions()
]
)
configured = self._async_configured_channel_ids()
options: list[SelectOptionDict] = []
seen: set[str] = set()
for option in selectable_channels:
if option["value"] in seen or option["value"] in configured:
continue
seen.add(option["value"])
options.append(option)
if not options:
return self.async_abort(reason="no_subscriptions")
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
probatio.Schema(
{
probatio.Required(CONF_CHANNELS): SelectSelector(
SelectSelectorConfig(
options=selectable_channels, multiple=True
)
),
}
),
self.config_entry.options,
step_id="user",
data_schema=probatio.Schema(
{
probatio.Required(CONF_CHANNEL_ID): SelectSelector(
SelectSelectorConfig(
options=options, mode=SelectSelectorMode.DROPDOWN
)
),
}
),
)
@callback
def _async_configured_channel_ids(self) -> set[str]:
"""Return channel ids already tracked by this config entry."""
return {
subentry.unique_id
for subentry in self._get_entry().get_subentries_of_type(
SUBENTRY_TYPE_CHANNEL
)
if subentry.unique_id
}
async def _async_get_youtube(self, config_entry: YouTubeConfigEntry) -> YouTube:
"""Return a YouTube client using a refreshed OAuth token."""
implementation = await async_get_config_entry_implementation(
self.hass, config_entry
)
auth = AsyncConfigEntryAuth(
self.hass, OAuth2Session(self.hass, config_entry, implementation)
)
return await auth.get_resource()
async def _async_create_entry(self, channel_id: str) -> SubentryFlowResult:
"""Create a subentry for the selected channel."""
config_entry: YouTubeConfigEntry = self._get_entry()
try:
youtube = await self._async_get_youtube(config_entry)
channels = [channel async for channel in youtube.get_channels([channel_id])]
except ForbiddenError as ex:
error = ex.args[0]
return self.async_abort(
reason="access_not_configured",
description_placeholders={"message": error},
)
except Exception as ex: # noqa: BLE001
LOGGER.error("Unknown error occurred: %s", ex.args)
return self.async_abort(reason="unknown")
if not channels or channels[0].snippet is None:
return self.async_abort(reason="unknown_channel")
return self.async_create_entry(
title=channels[0].snippet.title,
data={CONF_CHANNEL_ID: channel_id},
unique_id=channel_id,
)
homeassistant/components/youtube/const.py
+6
View File
@@ -7,9 +7,15 @@ DOMAIN = "youtube"
MANUFACTURER = "Google, Inc."
CHANNEL_CREATION_HELP_URL = "https://support.google.com/youtube/answer/1646861"
CONF_CHANNEL_ID = "channel_id"
CONF_CHANNELS = "channels"
CONF_UPLOAD_PLAYLIST = "upload_playlist_id"
SUBENTRY_TYPE_CHANNEL = "channel"
# The YouTube channels.list API accepts at most 50 ids per request
MAX_CHANNEL_IDS_PER_REQUEST = 50
LOGGER = logging.getLogger(__package__)
ATTR_TITLE = "title"
homeassistant/components/youtube/coordinator.py
+77 -52
View File
@@ -2,6 +2,7 @@
import asyncio
from datetime import timedelta
from itertools import batched
from typing import Any, Final, override
from youtubeaio.types import UnauthorizedError, YouTubeBackendError
@@ -26,9 +27,11 @@ from .const import (
ATTR_TOTAL_VIEWS,
ATTR_VIDEO_COUNT,
ATTR_VIDEO_ID,
CONF_CHANNELS,
CONF_CHANNEL_ID,
DOMAIN,
LOGGER,
MAX_CHANNEL_IDS_PER_REQUEST,
SUBENTRY_TYPE_CHANNEL,
)
type YouTubeConfigEntry = ConfigEntry[YouTubeDataUpdateCoordinator]
@@ -53,7 +56,11 @@ def _build_video_dict(video: Any, is_short: bool) -> dict[str, Any]:
class YouTubeDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""A YouTube Data Update Coordinator."""
"""A YouTube Data Update Coordinator fetching all tracked channels.
The data maps channel id to its channel data. Channels missing from
the API response are absent from the data.
"""
config_entry: YouTubeConfigEntry
@@ -66,6 +73,10 @@ class YouTubeDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Initialize the YouTube data coordinator."""
self._auth = auth
self._is_short_cache: dict[str, bool] = {}
self.subentry_ids = {
subentry.subentry_id
for subentry in config_entry.get_subentries_of_type(SUBENTRY_TYPE_CHANNEL)
}
super().__init__(
hass,
LOGGER,
@@ -76,65 +87,79 @@ class YouTubeDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
@override
async def _async_update_data(self) -> dict[str, Any]:
"""Fetch the data of all tracked channels in batched calls."""
youtube = await self._auth.get_resource()
res = {}
channel_ids = self.config_entry.options[CONF_CHANNELS]
channel_ids = [
subentry.data[CONF_CHANNEL_ID]
for subentry in self.config_entry.get_subentries_of_type(
SUBENTRY_TYPE_CHANNEL
)
]
if not channel_ids:
return {}
res: dict[str, Any] = {}
try:
async for channel in youtube.get_channels(channel_ids):
checked = 0
latest_video: dict[str, Any] | None = None
latest_short: dict[str, Any] | None = None
latest_video_non_short: dict[str, Any] | None = None
try:
async with asyncio.timeout(_SHORTS_DETECTION_TIMEOUT):
# Only examine the first page (10 items): paginating
# further burns API quota for no benefit.
async for video in youtube.get_playlist_items(
channel.upload_playlist_id, 10
):
checked += 1
is_short = await self._resolve_is_short(youtube, video)
entry = _build_video_dict(video, is_short)
if latest_video is None:
latest_video = entry
if is_short and latest_short is None:
latest_short = entry
if not is_short and latest_video_non_short is None:
latest_video_non_short = entry
if (
latest_short is not None
and latest_video_non_short is not None
):
break
if checked >= 10:
break
except TimeoutError:
LOGGER.warning(
"Timed out processing recent uploads for channel %s; "
"continuing with partial results",
channel.channel_id,
for chunk in batched(
channel_ids, MAX_CHANNEL_IDS_PER_REQUEST, strict=False
):
async for channel in youtube.get_channels(list(chunk)):
res[channel.channel_id] = await self._async_get_channel_data(
youtube, channel
)
LOGGER.debug(
"Examined %d videos for channel %s", checked, channel.channel_id
)
res[channel.channel_id] = {
ATTR_ID: channel.channel_id,
ATTR_TITLE: channel.snippet.title,
ATTR_ICON: channel.snippet.thumbnails.get_highest_quality().url,
ATTR_LATEST_VIDEO: latest_video,
ATTR_LATEST_SHORT: latest_short,
ATTR_LATEST_VIDEO_NON_SHORT: latest_video_non_short,
ATTR_SUBSCRIBER_COUNT: channel.statistics.subscriber_count,
ATTR_TOTAL_VIEWS: channel.statistics.view_count,
ATTR_VIDEO_COUNT: channel.statistics.video_count,
}
except UnauthorizedError as err:
raise ConfigEntryAuthFailed from err
except YouTubeBackendError as err:
raise UpdateFailed("Couldn't connect to YouTube") from err
return res
async def _async_get_channel_data(
self, youtube: Any, channel: Any
) -> dict[str, Any]:
"""Fetch the uploads of the channel and build its sensor data."""
checked = 0
latest_video: dict[str, Any] | None = None
latest_short: dict[str, Any] | None = None
latest_video_non_short: dict[str, Any] | None = None
try:
async with asyncio.timeout(_SHORTS_DETECTION_TIMEOUT):
# Only examine the first page (10 items): paginating
# further burns API quota for no benefit.
async for video in youtube.get_playlist_items(
channel.upload_playlist_id, 10
):
checked += 1
is_short = await self._resolve_is_short(youtube, video)
entry = _build_video_dict(video, is_short)
if latest_video is None:
latest_video = entry
if is_short and latest_short is None:
latest_short = entry
if not is_short and latest_video_non_short is None:
latest_video_non_short = entry
if latest_short is not None and latest_video_non_short is not None:
break
if checked >= 10:
break
except TimeoutError:
LOGGER.warning(
"Timed out processing recent uploads for channel %s; "
"continuing with partial results",
channel.channel_id,
)
LOGGER.debug("Examined %d videos for channel %s", checked, channel.channel_id)
return {
ATTR_ID: channel.channel_id,
ATTR_TITLE: channel.snippet.title,
ATTR_ICON: channel.snippet.thumbnails.get_highest_quality().url,
ATTR_LATEST_VIDEO: latest_video,
ATTR_LATEST_SHORT: latest_short,
ATTR_LATEST_VIDEO_NON_SHORT: latest_video_non_short,
ATTR_SUBSCRIBER_COUNT: channel.statistics.subscriber_count,
ATTR_TOTAL_VIEWS: channel.statistics.view_count,
ATTR_VIDEO_COUNT: channel.statistics.video_count,
}
async def _resolve_is_short(self, youtube: Any, video: Any) -> bool:
"""Return whether a single video is a Short.
homeassistant/components/youtube/diagnostics.py
+1 -2
View File
@@ -17,9 +17,8 @@ async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: YouTubeConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator = entry.runtime_data
sensor_data: dict[str, Any] = {}
for channel_id, channel_data in coordinator.data.items():
for channel_id, channel_data in entry.runtime_data.data.items():
channel_copy = dict(channel_data)
# Strip verbose description field from all video entries.
for attr in (
homeassistant/components/youtube/entity.py
+23 -6
View File
@@ -1,10 +1,13 @@
"""Entity representing a YouTube account."""
"""Entity representing a YouTube channel."""
from typing import Any, override
from homeassistant.config_entries import ConfigSubentry
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTR_TITLE, DOMAIN, MANUFACTURER
from .const import CONF_CHANNEL_ID, DOMAIN, MANUFACTURER
from .coordinator import YouTubeDataUpdateCoordinator
@@ -16,19 +19,33 @@ class YouTubeChannelEntity(CoordinatorEntity[YouTubeDataUpdateCoordinator]):
def __init__(
self,
coordinator: YouTubeDataUpdateCoordinator,
subentry: ConfigSubentry,
description: EntityDescription,
channel_id: str,
) -> None:
"""Initialize a YouTube entity."""
super().__init__(coordinator)
self.entity_description = description
channel_id = subentry.data[CONF_CHANNEL_ID]
self._channel_id = channel_id
# The entry id prefix keeps unique ids unique when two accounts
# track the same channel.
self._attr_unique_id = (
f"{coordinator.config_entry.entry_id}_{channel_id}_{description.key}"
)
self._channel_id = channel_id
self._attr_device_info = DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, f"{coordinator.config_entry.entry_id}_{channel_id}")},
identifiers={(DOMAIN, channel_id)},
manufacturer=MANUFACTURER,
name=coordinator.data[channel_id][ATTR_TITLE],
name=subentry.title,
)
@property
@override
def available(self) -> bool:
"""Return if the entity is available."""
return super().available and self._channel_id in self.coordinator.data
@property
def _channel_data(self) -> Any:
"""Return the channel data."""
return self.coordinator.data[self._channel_id]
homeassistant/components/youtube/sensor.py
+13 -13
View File
@@ -25,6 +25,7 @@ from .const import (
ATTR_TOTAL_VIEWS,
ATTR_VIDEO_COUNT,
ATTR_VIDEO_ID,
SUBENTRY_TYPE_CHANNEL,
)
from .coordinator import YouTubeConfigEntry
from .entity import YouTubeChannelEntity
@@ -117,11 +118,14 @@ async def async_setup_entry(
) -> None:
"""Set up the YouTube sensor."""
coordinator = entry.runtime_data
async_add_entities(
YouTubeSensor(coordinator, sensor_type, channel_id)
for channel_id in coordinator.data
for sensor_type in SENSOR_TYPES
)
for subentry in entry.get_subentries_of_type(SUBENTRY_TYPE_CHANNEL):
async_add_entities(
(
YouTubeSensor(coordinator, subentry, sensor_type)
for sensor_type in SENSOR_TYPES
),
config_subentry_id=subentry.subentry_id,
)
class YouTubeSensor(YouTubeChannelEntity, SensorEntity):
@@ -134,14 +138,14 @@ class YouTubeSensor(YouTubeChannelEntity, SensorEntity):
def available(self) -> bool:
"""Return if the entity is available."""
return super().available and self.entity_description.available_fn(
self.coordinator.data[self._channel_id]
self._channel_data
)
@property
@override
def native_value(self) -> StateType:
"""Return the value reported by the sensor."""
return self.entity_description.value_fn(self.coordinator.data[self._channel_id])
return self.entity_description.value_fn(self._channel_data)
@property
@override
@@ -149,16 +153,12 @@ class YouTubeSensor(YouTubeChannelEntity, SensorEntity):
"""Return the value reported by the sensor."""
if not self.available:
return None
return self.entity_description.entity_picture_fn(
self.coordinator.data[self._channel_id]
)
return self.entity_description.entity_picture_fn(self._channel_data)
@property
@override
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the extra state attributes."""
if self.entity_description.attributes_fn:
return self.entity_description.attributes_fn(
self.coordinator.data[self._channel_id]
)
return self.entity_description.attributes_fn(self._channel_data)
return None
homeassistant/components/youtube/strings.json
+21 -10
View File
@@ -26,6 +26,27 @@
}
}
},
"config_subentries": {
"channel": {
"abort": {
"access_not_configured": "[%key:component::youtube::config::abort::access_not_configured%]",
"already_configured": "This channel is already configured.",
"no_subscriptions": "There are no channels available to add.",
"unknown": "[%key:common::config_flow::error::unknown%]",
"unknown_channel": "The selected channel is no longer available."
},
"entry_type": "YouTube channel",
"initiate_flow": {
"user": "Add channel"
},
"step": {
"user": {
"data": { "channel_id": "Channel" },
"description": "Select the channel you want to add."
}
}
}
},
"entity": {
"sensor": {
"latest_short": {
@@ -53,15 +74,5 @@
"videos": { "name": "Videos" },
"views": { "name": "Views" }
}
},
"options": {
"step": {
"init": {
"data": {
"channels": "[%key:component::youtube::config::step::channels::data::channels%]"
},
"description": "[%key:component::youtube::config::step::channels::description%]"
}
}
}
}
tests/components/youtube/__init__.py
+7 -5
View File
@@ -24,6 +24,7 @@ class MockYouTube:
playlist_items_fixture: str = "get_playlist_items.json",
subscriptions_fixture: str = "get_subscriptions.json",
short_video_ids: set[str] | None = None,
extra_channel_fixtures: list[str] | None = None,
short_check_delay: float = 0.0,
) -> None:
"""Initialize mock service."""
@@ -32,6 +33,7 @@ class MockYouTube:
self._playlist_items_fixture = playlist_items_fixture
self._subscriptions_fixture = subscriptions_fixture
self._short_video_ids: set[str] = short_video_ids or set()
self._extra_channel_fixtures = extra_channel_fixtures or []
self._short_check_delay = short_check_delay
self.playlist_item_requests = 0
self.playlist_items_yielded = 0
@@ -55,11 +57,11 @@ class MockYouTube:
"""Get channels."""
if self._thrown_error is not None:
raise self._thrown_error
channels = await async_load_json_object_fixture(
self.hass, self._channel_fixture, DOMAIN
)
for item in channels["items"]:
yield YouTubeChannel(**item)
for fixture in (self._channel_fixture, *self._extra_channel_fixtures):
channels = await async_load_json_object_fixture(self.hass, fixture, DOMAIN)
for item in channels["items"]:
if item["id"] in channel_ids:
yield YouTubeChannel(**item)
async def get_playlist_items(
self, playlist_id: str, amount: int
tests/components/youtube/conftest.py
+33 -12
View File
@@ -12,7 +12,12 @@ from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.youtube.const import DOMAIN
from homeassistant.components.youtube.const import (
CONF_CHANNEL_ID,
DOMAIN,
SUBENTRY_TYPE_CHANNEL,
)
from homeassistant.config_entries import ConfigSubentryData
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@@ -31,12 +36,27 @@ SCOPES = [
"https://www.googleapis.com/auth/youtube.readonly",
]
TITLE = "Google for Developers"
CHANNEL_ID = "UC_x5XG1OV2P6uZZ5FSM9Ttw"
LINUS_CHANNEL_ID = "UCXuqSBlHAE6Xw-yeJA0Tunw"
TOKEN = (
"homeassistant.components.youtube.api"
".config_entry_oauth2_flow.OAuth2Session.async_ensure_token_valid"
)
def mock_entry_data(expires_at: int, scopes: list[str]) -> dict[str, Any]:
"""Return OAuth data for a YouTube config entry."""
return {
"auth_implementation": DOMAIN,
"token": {
"access_token": "mock-access-token",
"refresh_token": "mock-refresh-token",
"expires_at": expires_at,
"scope": " ".join(scopes),
},
}
@pytest.fixture(name="scopes")
def mock_scopes() -> list[str]:
"""Fixture to set the scopes present in the OAuth token."""
@@ -67,17 +87,18 @@ def mock_config_entry(expires_at: int, scopes: list[str]) -> MockConfigEntry:
return MockConfigEntry(
domain=DOMAIN,
title=TITLE,
unique_id="UC_x5XG1OV2P6uZZ5FSM9Ttw",
data={
"auth_implementation": DOMAIN,
"token": {
"access_token": "mock-access-token",
"refresh_token": "mock-refresh-token",
"expires_at": expires_at,
"scope": " ".join(scopes),
},
},
options={"channels": ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]},
unique_id=CHANNEL_ID,
version=2,
data=mock_entry_data(expires_at, scopes),
subentries_data=[
ConfigSubentryData(
data={CONF_CHANNEL_ID: CHANNEL_ID},
subentry_id="channel_1",
subentry_type=SUBENTRY_TYPE_CHANNEL,
title="Google for Developers",
unique_id=CHANNEL_ID,
)
],
)
tests/components/youtube/test_config_flow.py
+381 -68
View File
@@ -1,21 +1,30 @@
"""Test the YouTube config flow."""
import time
from unittest.mock import patch
import pytest
from youtubeaio.types import ForbiddenError
from homeassistant import config_entries
from homeassistant.components.youtube.const import CONF_CHANNELS, DOMAIN
from homeassistant.components.youtube.const import (
CONF_CHANNEL_ID,
CONF_CHANNELS,
DOMAIN,
SUBENTRY_TYPE_CHANNEL,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.exceptions import OAuth2TokenRequestConnectionError
from homeassistant.helpers import config_entry_oauth2_flow
from . import MockYouTube
from .conftest import (
CHANNEL_ID,
CLIENT_ID,
GOOGLE_AUTH_URI,
GOOGLE_TOKEN_URI,
LINUS_CHANNEL_ID,
SCOPES,
TITLE,
ComponentSetup,
@@ -69,7 +78,7 @@ async def test_full_flow(
assert result["step_id"] == "channels"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]}
result["flow_id"], user_input={CONF_CHANNELS: [CHANNEL_ID]}
)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
@@ -78,11 +87,19 @@ async def test_full_flow(
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TITLE
assert "result" in result
assert result["result"].unique_id == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
assert "token" in result["result"].data
assert result["result"].data["token"]["access_token"] == "mock-access-token"
assert result["result"].data["token"]["refresh_token"] == "mock-refresh-token"
assert result["options"] == {CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]}
entry = result["result"]
assert entry.unique_id == CHANNEL_ID
assert entry.version == 2
assert "token" in entry.data
assert entry.data["token"]["access_token"] == "mock-access-token"
assert entry.data["token"]["refresh_token"] == "mock-refresh-token"
assert entry.options == {}
assert len(entry.subentries) == 1
subentry = next(iter(entry.subentries.values()))
assert subentry.subentry_type == SUBENTRY_TYPE_CHANNEL
assert subentry.unique_id == CHANNEL_ID
assert subentry.title == "Google for Developers"
assert subentry.data == {CONF_CHANNEL_ID: CHANNEL_ID}
@pytest.mark.usefixtures("current_request_with_host")
@@ -215,23 +232,27 @@ async def test_flow_without_subscriptions(
schema = result["data_schema"]
channels = schema.schema[CONF_CHANNELS].config["options"]
assert len(channels) == 1
assert channels[0]["value"] == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
assert channels[0]["value"] == CHANNEL_ID
assert "(Your Channel)" in channels[0]["label"]
# Test selecting the own channel
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]},
user_input={CONF_CHANNELS: [CHANNEL_ID]},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TITLE
assert "result" in result
assert result["result"].unique_id == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
assert "token" in result["result"].data
assert result["result"].data["token"]["access_token"] == "mock-access-token"
assert result["result"].data["token"]["refresh_token"] == "mock-refresh-token"
assert result["options"] == {CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]}
entry = result["result"]
assert entry.unique_id == CHANNEL_ID
assert "token" in entry.data
assert entry.data["token"]["access_token"] == "mock-access-token"
assert entry.data["token"]["refresh_token"] == "mock-refresh-token"
assert entry.options == {}
assert len(entry.subentries) == 1
subentry = next(iter(entry.subentries.values()))
assert subentry.unique_id == CHANNEL_ID
@pytest.mark.usefixtures("current_request_with_host")
@@ -384,7 +405,7 @@ async def test_reauth(
assert result["description_placeholders"] == placeholders
assert len(mock_setup.mock_calls) == call_count
assert config_entry.unique_id == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
assert config_entry.unique_id == CHANNEL_ID
assert "token" in config_entry.data
# Verify access token is refreshed
assert config_entry.data["token"]["access_token"] == access_token
@@ -428,30 +449,70 @@ async def test_flow_exception(
assert result["reason"] == "unknown"
async def test_options_flow(
hass: HomeAssistant, setup_integration: ComponentSetup
@pytest.mark.parametrize(
("exception", "abort_reason", "placeholders"),
[
(
ForbiddenError(
"YouTube Data API v3 has not been used in project 0"
" before or it is disabled."
),
"access_not_configured",
{
"message": "YouTube Data API v3 has not been used in project 0"
" before or it is disabled."
},
),
(Exception("Some failure"), "unknown", None),
],
ids=["forbidden", "unknown"],
)
@pytest.mark.usefixtures("current_request_with_host")
async def test_flow_channel_listing_error(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
exception: Exception,
abort_reason: str,
placeholders: dict[str, str] | None,
) -> None:
"""Test the full options flow."""
await setup_integration()
with patch(
"homeassistant.components.youtube.config_flow.YouTube",
return_value=MockYouTube(hass),
"""Test the initial flow aborts when listing the channels fails."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["url"] == (
f"{GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope={'+'.join(SCOPES)}"
"&access_type=offline&prompt=consent"
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
# The account check succeeds, listing the subscriptions afterwards fails
mock = MockYouTube(hass)
with (
patch("homeassistant.components.youtube.async_setup_entry", return_value=True),
patch(
"homeassistant.components.youtube.config_flow.YouTube",
return_value=mock,
),
patch.object(mock, "get_user_subscriptions", side_effect=exception),
):
entry = hass.config_entries.async_entries(DOMAIN)[0]
result = await hass.config_entries.options.async_init(entry.entry_id)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]}
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == abort_reason
assert result.get("description_placeholders") == placeholders
@pytest.mark.usefixtures("current_request_with_host")
@@ -500,17 +561,14 @@ async def test_own_channel_included(
schema = result["data_schema"]
channels = schema.schema[CONF_CHANNELS].config["options"]
assert any(
channel["value"] == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
and "(Your Channel)" in channel["label"]
channel["value"] == CHANNEL_ID and "(Your Channel)" in channel["label"]
for channel in channels
)
# Test selecting both own channel and a subscribed channel
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw", "UC_x5XG1OV2P6uZZ5FSM9Ttw"]
},
user_input={CONF_CHANNELS: [CHANNEL_ID, CHANNEL_ID]},
)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
@@ -519,45 +577,300 @@ async def test_own_channel_included(
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TITLE
assert "result" in result
assert result["result"].unique_id == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
assert "token" in result["result"].data
assert result["result"].data["token"]["access_token"] == "mock-access-token"
assert result["result"].data["token"]["refresh_token"] == "mock-refresh-token"
assert result["options"] == {
CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw", "UC_x5XG1OV2P6uZZ5FSM9Ttw"]
}
entry = result["result"]
assert entry.unique_id == CHANNEL_ID
assert "token" in entry.data
assert entry.data["token"]["access_token"] == "mock-access-token"
assert entry.data["token"]["refresh_token"] == "mock-refresh-token"
assert entry.options == {}
# Duplicate selections are deduplicated into a single subentry
assert len(entry.subentries) == 1
subentry = next(iter(entry.subentries.values()))
assert subentry.unique_id == CHANNEL_ID
assert subentry.title == "Google for Developers"
async def test_options_flow_own_channel(
async def test_subentry_flow_add_channel(
hass: HomeAssistant, setup_integration: ComponentSetup
) -> None:
"""Test the options flow includes the user's own channel."""
"""Test adding a channel subentry."""
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
# Listing shows the Linus channel, the Google channel is already tracked;
# the reload after adding needs both channels
mock = MockYouTube(
hass,
channel_fixture="get_channel_2.json",
extra_channel_fixtures=["get_channel.json"],
)
with patch(
"homeassistant.components.youtube.config_flow.YouTube",
"homeassistant.components.youtube.api.YouTube",
return_value=mock,
):
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_CHANNEL),
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
# The already configured channel is not selectable
options = result["data_schema"].schema[CONF_CHANNEL_ID].config["options"]
assert [option["value"] for option in options] == [LINUS_CHANNEL_ID]
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], user_input={CONF_CHANNEL_ID: LINUS_CHANNEL_ID}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Linus Tech Tips"
assert result["data"] == {CONF_CHANNEL_ID: LINUS_CHANNEL_ID}
assert len(entry.subentries) == 2
new_subentry = next(
subentry
for subentry in entry.subentries.values()
if subentry.unique_id == LINUS_CHANNEL_ID
)
assert new_subentry.title == "Linus Tech Tips"
assert hass.states.get("sensor.linus_tech_tips_subscribers") is not None
async def test_subentry_flow_no_channels_left(
hass: HomeAssistant, setup_integration: ComponentSetup
) -> None:
"""Test the subentry flow aborts when all channels are already tracked."""
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
with patch(
"homeassistant.components.youtube.api.YouTube",
return_value=MockYouTube(hass),
):
entry = hass.config_entries.async_entries(DOMAIN)[0]
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_CHANNEL),
context={"source": config_entries.SOURCE_USER},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_subscriptions"
async def test_subentry_flow_add_own_channel(
hass: HomeAssistant, setup_integration: ComponentSetup
) -> None:
"""Test the account's own channel can be added if not tracked yet."""
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
hass.config_entries.async_remove_subentry(entry, "channel_1")
await hass.async_block_till_done()
assert not entry.subentries
with patch(
"homeassistant.components.youtube.api.YouTube",
return_value=MockYouTube(hass),
):
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_CHANNEL),
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
assert result["step_id"] == "user"
# Verify the form schema contains the user's own channel
schema = result["data_schema"]
channels = schema.schema[CONF_CHANNELS].config["options"]
assert any(
channel["value"] == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
and "(Your Channel)" in channel["label"]
for channel in channels
)
options = result["data_schema"].schema[CONF_CHANNEL_ID].config["options"]
assert [option["value"] for option in options] == [CHANNEL_ID]
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]},
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], user_input={CONF_CHANNEL_ID: CHANNEL_ID}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]}
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Google for Developers"
assert len(entry.subentries) == 1
subentry = next(iter(entry.subentries.values()))
assert subentry.unique_id == CHANNEL_ID
assert subentry.title == "Google for Developers"
assert hass.states.get("sensor.google_for_developers_subscribers") is not None
async def test_subentry_flow_unknown_channel(
hass: HomeAssistant, setup_integration: ComponentSetup
) -> None:
"""Test the subentry flow aborts when the channel is not available."""
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
hass.config_entries.async_remove_subentry(entry, "channel_1")
await hass.async_block_till_done()
# The subscription lists the channel, but it cannot be fetched anymore
mock = MockYouTube(hass, channel_fixture="get_no_channel.json")
with patch(
"homeassistant.components.youtube.api.YouTube",
return_value=mock,
):
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_CHANNEL),
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], user_input={CONF_CHANNEL_ID: CHANNEL_ID}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unknown_channel"
assert not entry.subentries
assert entry.state is config_entries.ConfigEntryState.LOADED
@pytest.mark.parametrize(
("exception", "abort_reason", "placeholders"),
[
(
ForbiddenError(
"YouTube Data API v3 has not been used in project 0"
" before or it is disabled."
),
"access_not_configured",
{
"message": "YouTube Data API v3 has not been used in project 0"
" before or it is disabled."
},
),
(Exception("Some failure"), "unknown", None),
],
ids=["forbidden", "unknown"],
)
@pytest.mark.usefixtures("current_request_with_host")
async def test_subentry_flow_api_error_listing_channels(
hass: HomeAssistant,
setup_integration: ComponentSetup,
exception: Exception,
abort_reason: str,
placeholders: dict[str, str] | None,
) -> None:
"""Test the subentry flow aborts when listing channels fails."""
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
mock = MockYouTube(hass)
with (
patch(
"homeassistant.components.youtube.api.YouTube",
return_value=mock,
),
patch.object(mock, "get_user_channels", side_effect=exception),
):
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_CHANNEL),
context={"source": config_entries.SOURCE_USER},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == abort_reason
assert result.get("description_placeholders") == placeholders
@pytest.mark.parametrize(
("exception", "abort_reason", "placeholders"),
[
(
ForbiddenError(
"YouTube Data API v3 has not been used in project 0"
" before or it is disabled."
),
"access_not_configured",
{
"message": "YouTube Data API v3 has not been used in project 0"
" before or it is disabled."
},
),
(Exception("Some failure"), "unknown", None),
],
ids=["forbidden", "unknown"],
)
async def test_subentry_flow_api_error_fetching_channel(
hass: HomeAssistant,
setup_integration: ComponentSetup,
exception: Exception,
abort_reason: str,
placeholders: dict[str, str] | None,
) -> None:
"""Test the subentry flow aborts when fetching the selected channel fails."""
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
hass.config_entries.async_remove_subentry(entry, "channel_1")
await hass.async_block_till_done()
# Listing channels for the form succeeds, only get_channels raises
mock = MockYouTube(hass)
mock.set_thrown_exception(exception)
with patch(
"homeassistant.components.youtube.api.YouTube",
return_value=mock,
):
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_CHANNEL),
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], user_input={CONF_CHANNEL_ID: CHANNEL_ID}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == abort_reason
assert result.get("description_placeholders") == placeholders
async def test_subentry_flow_token_refresh_error(
hass: HomeAssistant, setup_integration: ComponentSetup
) -> None:
"""Test the subentry flow aborts when the token cannot be refreshed."""
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
hass.config_entries.async_remove_subentry(entry, "channel_1")
await hass.async_block_till_done()
with patch(
"homeassistant.components.youtube.api.YouTube",
return_value=MockYouTube(hass),
):
# Listing channels succeeds with the still valid token
result = await hass.config_entries.subentries.async_init(
(entry.entry_id, SUBENTRY_TYPE_CHANNEL),
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
# The token expires before the submission and the refresh fails
hass.config_entries.async_update_entry(
entry,
data={
**entry.data,
"token": {
**entry.data["token"],
"expires_at": time.time() - 3600,
},
},
)
await hass.async_block_till_done()
with patch(
"homeassistant.components.youtube.OAuth2Session.async_ensure_token_valid",
side_effect=OAuth2TokenRequestConnectionError(domain=DOMAIN),
):
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], user_input={CONF_CHANNEL_ID: CHANNEL_ID}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unknown"
tests/components/youtube/test_init.py
+368 -7
View File
@@ -1,22 +1,41 @@
"""Tests for YouTube."""
from collections.abc import AsyncGenerator
import http
import time
from unittest.mock import patch
import pytest
from youtubeaio.models import YouTubeChannel
from homeassistant.components.youtube.const import CONF_CHANNELS, DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.components.youtube.const import (
CONF_CHANNEL_ID,
CONF_CHANNELS,
DOMAIN,
SUBENTRY_TYPE_CHANNEL,
)
from homeassistant.components.youtube.diagnostics import (
async_get_config_entry_diagnostics,
)
from homeassistant.config_entries import ConfigEntryState, ConfigSubentryData
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import OAuth2TokenRequestConnectionError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.config_entry_oauth2_flow import (
ImplementationUnavailableError,
)
from .conftest import GOOGLE_TOKEN_URI, ComponentSetup
from . import MockYouTube
from .conftest import (
CHANNEL_ID,
GOOGLE_TOKEN_URI,
LINUS_CHANNEL_ID,
TITLE,
ComponentSetup,
mock_entry_data,
)
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
@@ -128,17 +147,359 @@ async def test_device_info(
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
channel_id = entry.options[CONF_CHANNELS][0]
device = device_registry.async_get_device_by_identifier(
(DOMAIN, f"{entry.entry_id}_{channel_id}"), entry.entry_id
(DOMAIN, CHANNEL_ID), entry.entry_id
)
assert device.entry_type is dr.DeviceEntryType.SERVICE
assert device.identifiers == {(DOMAIN, f"{entry.entry_id}_{channel_id}")}
assert device.identifiers == {(DOMAIN, CHANNEL_ID)}
assert device.manufacturer == "Google, Inc."
assert device.name == "Google for Developers"
async def test_migration(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
expires_at: int,
scopes: list[str],
) -> None:
"""Test migration of an options based entry to the subentry structure."""
entry = MockConfigEntry(
domain=DOMAIN,
title=TITLE,
unique_id=CHANNEL_ID,
version=1,
data=mock_entry_data(expires_at, scopes),
options={CONF_CHANNELS: [CHANNEL_ID]},
)
entry.add_to_hass(hass)
device = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
entry_type=dr.DeviceEntryType.SERVICE,
identifiers={(DOMAIN, f"{entry.entry_id}_{CHANNEL_ID}")},
manufacturer="Google, Inc.",
# Stale name: the channel was renamed since the device was created
name="Google Developers Channel",
)
entity_registry.async_get_or_create(
"sensor",
DOMAIN,
f"{entry.entry_id}_{CHANNEL_ID}_subscribers",
config_entry=entry,
device_id=device.id,
suggested_object_id="google_for_developers_subscribers",
)
# Device and entity left behind by a channel which is no longer tracked,
# as happened when it was removed from the options before this migration.
orphan_device = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
entry_type=dr.DeviceEntryType.SERVICE,
identifiers={(DOMAIN, f"{entry.entry_id}_{LINUS_CHANNEL_ID}")},
manufacturer="Google, Inc.",
name="Linus Tech Tips",
)
entity_registry.async_get_or_create(
"sensor",
DOMAIN,
f"{entry.entry_id}_{LINUS_CHANNEL_ID}_subscribers",
config_entry=entry,
device_id=orphan_device.id,
suggested_object_id="linus_tech_tips_subscribers",
)
with patch(
"homeassistant.components.youtube.api.YouTube",
return_value=MockYouTube(hass),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert entry.version == 2
assert entry.options == {}
assert len(entry.subentries) == 1
subentry = next(iter(entry.subentries.values()))
assert subentry.subentry_type == SUBENTRY_TYPE_CHANNEL
assert subentry.unique_id == CHANNEL_ID
# Entry titles are userland: the migrated subentry keeps the device
# name, even though the channel was renamed on YouTube since.
assert subentry.title == "Google Developers Channel"
assert subentry.data == {CONF_CHANNEL_ID: CHANNEL_ID}
migrated_device = device_registry.async_get_device_by_identifier(
(DOMAIN, CHANNEL_ID), entry.entry_id
)
assert migrated_device is not None
assert migrated_device.id == device.id
assert migrated_device.config_subentry_id == subentry.subentry_id
# The unique id keeps the entry id prefix so two accounts can track
# the same channel.
migrated_entity_id = entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{entry.entry_id}_{CHANNEL_ID}_subscribers"
)
assert migrated_entity_id is not None
migrated_entity = entity_registry.async_get(migrated_entity_id)
assert migrated_entity is not None
assert migrated_entity.config_subentry_id == subentry.subentry_id
assert migrated_entity.device_id == device.id
# The untracked channel's device and entity are cleaned up
assert device_registry.async_get(orphan_device.id) is None
assert (
entity_registry.async_get_entity_id(
"sensor", DOMAIN, f"{entry.entry_id}_{LINUS_CHANNEL_ID}_subscribers"
)
is None
)
assert hass.states.get("sensor.google_for_developers_subscribers") is not None
async def test_remove_subentry(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
setup_integration: ComponentSetup,
) -> None:
"""Test removing a channel subentry removes its device and entities."""
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
subentry = next(iter(entry.subentries.values()))
device = device_registry.async_get_device_by_identifier(
(DOMAIN, CHANNEL_ID), entry.entry_id
)
assert device is not None
hass.config_entries.async_remove_subentry(entry, subentry.subentry_id)
await hass.async_block_till_done()
assert not entry.subentries
assert (
device_registry.async_get_device_by_identifier(
(DOMAIN, CHANNEL_ID), entry.entry_id
)
is None
)
assert hass.states.get("sensor.google_for_developers_subscribers") is None
async def test_channel_missing_from_api(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
setup_integration: ComponentSetup,
) -> None:
"""Test the channel's entities are unavailable when the API omits it."""
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
device = device_registry.async_get_device_by_identifier(
(DOMAIN, CHANNEL_ID), entry.entry_id
)
assert device is not None
with patch(
"homeassistant.components.youtube.api.AsyncConfigEntryAuth.get_resource",
return_value=MockYouTube(hass, channel_fixture="get_no_channel.json"),
):
await hass.config_entries.async_reload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
state = hass.states.get("sensor.google_for_developers_subscribers")
assert state is not None
assert state.state == "unavailable"
assert (
device_registry.async_get_device_by_identifier(
(DOMAIN, CHANNEL_ID), entry.entry_id
)
is not None
)
async def test_missing_channel_does_not_affect_other_channels(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
expires_at: int,
scopes: list[str],
) -> None:
"""Test a channel missing from the API only takes itself down."""
entry = MockConfigEntry(
domain=DOMAIN,
title=TITLE,
unique_id=CHANNEL_ID,
version=2,
data=mock_entry_data(expires_at, scopes),
subentries_data=[
ConfigSubentryData(
data={CONF_CHANNEL_ID: CHANNEL_ID},
subentry_id="channel_1",
subentry_type=SUBENTRY_TYPE_CHANNEL,
title="Google for Developers",
unique_id=CHANNEL_ID,
),
ConfigSubentryData(
data={CONF_CHANNEL_ID: LINUS_CHANNEL_ID},
subentry_id="channel_2",
subentry_type=SUBENTRY_TYPE_CHANNEL,
title="Linus Tech Tips",
unique_id=LINUS_CHANNEL_ID,
),
],
)
entry.add_to_hass(hass)
# The API only returns the Google channel, the Linus channel is missing
with patch(
"homeassistant.components.youtube.api.YouTube",
return_value=MockYouTube(hass),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
state = hass.states.get("sensor.google_for_developers_subscribers")
assert state is not None
assert state.state == "2290000"
state = hass.states.get("sensor.linus_tech_tips_subscribers")
assert state is not None
assert state.state == "unavailable"
assert (
device_registry.async_get_device_by_identifier(
(DOMAIN, CHANNEL_ID), entry.entry_id
)
is not None
)
assert (
device_registry.async_get_device_by_identifier(
(DOMAIN, LINUS_CHANNEL_ID), entry.entry_id
)
is not None
)
diagnostics = await async_get_config_entry_diagnostics(hass, entry)
assert diagnostics[CHANNEL_ID]["title"] == "Google for Developers"
assert LINUS_CHANNEL_ID not in diagnostics
async def test_more_channels_than_the_api_limit_are_chunked(
hass: HomeAssistant,
expires_at: int,
scopes: list[str],
) -> None:
"""Test channel ids are fetched in chunks of at most 50."""
channel_ids = [f"UC_channel_{i}" for i in range(51)]
entry = MockConfigEntry(
domain=DOMAIN,
title=TITLE,
unique_id=CHANNEL_ID,
version=2,
data=mock_entry_data(expires_at, scopes),
subentries_data=[
ConfigSubentryData(
data={CONF_CHANNEL_ID: channel_id},
subentry_id=f"channel_{i}",
subentry_type=SUBENTRY_TYPE_CHANNEL,
title=f"Channel {i}",
unique_id=channel_id,
)
for i, channel_id in enumerate(channel_ids)
],
)
entry.add_to_hass(hass)
mock = MockYouTube(hass)
requested_id_chunks: list[list[str]] = []
mock_get_channels = mock.get_channels
async def get_channels(channel_ids: list[str]) -> AsyncGenerator[YouTubeChannel]:
requested_id_chunks.append(list(channel_ids))
async for channel in mock_get_channels(channel_ids):
yield channel
mock.get_channels = get_channels
# None of the channels is known to the API
with patch(
"homeassistant.components.youtube.api.YouTube",
return_value=mock,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert requested_id_chunks == [channel_ids[:50], channel_ids[50:]]
assert entry.runtime_data.data == {}
state = hass.states.get("sensor.channel_0_subscribers")
assert state is not None
assert state.state == "unavailable"
async def test_migration_channel_missing_from_api(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
expires_at: int,
scopes: list[str],
) -> None:
"""Test a migrated channel keeps its name when it is not available."""
entry = MockConfigEntry(
domain=DOMAIN,
title=TITLE,
unique_id=CHANNEL_ID,
version=1,
data=mock_entry_data(expires_at, scopes),
options={CONF_CHANNELS: [CHANNEL_ID]},
)
entry.add_to_hass(hass)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
entry_type=dr.DeviceEntryType.SERVICE,
identifiers={(DOMAIN, f"{entry.entry_id}_{CHANNEL_ID}")},
manufacturer="Google, Inc.",
name="Google for Developers",
)
with patch(
"homeassistant.components.youtube.api.YouTube",
return_value=MockYouTube(hass, channel_fixture="get_no_channel.json"),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert len(entry.subentries) == 1
subentry = next(iter(entry.subentries.values()))
assert subentry.title == "Google for Developers"
state = hass.states.get("sensor.google_for_developers_subscribers")
assert state is not None
assert state.state == "unavailable"
device = device_registry.async_get_device_by_identifier(
(DOMAIN, CHANNEL_ID), entry.entry_id
)
assert device is not None
assert device.name == "Google for Developers"
async def test_entry_data_update_does_not_reload(
hass: HomeAssistant, setup_integration: ComponentSetup
) -> None:
"""Test token refreshes and other entry data updates do not reload."""
await setup_integration()
entry = hass.config_entries.async_entries(DOMAIN)[0]
coordinator = entry.runtime_data
hass.config_entries.async_update_entry(
entry,
data={
**entry.data,
"token": {**entry.data["token"], "access_token": "updated-access-token"},
},
)
await hass.async_block_till_done()
assert entry.runtime_data is coordinator
async def test_oauth_implementation_not_available(
hass: HomeAssistant, setup_integration: ComponentSetup
) -> None: