From 1aab95f9b5692510198476d7dab01be93b3f93c4 Mon Sep 17 00:00:00 2001 From: DevHugo Date: Mon, 14 Sep 2026 21:37:04 +0200 Subject: [PATCH] Fix youtube stream playlist items and stop early to avoid paginating entire playlists (#181321) --- .../components/youtube/coordinator.py | 106 ++++++++++-------- tests/components/youtube/__init__.py | 26 ++++- tests/components/youtube/test_sensor.py | 94 ++++++++++++++-- 3 files changed, 168 insertions(+), 58 deletions(-) diff --git a/homeassistant/components/youtube/coordinator.py b/homeassistant/components/youtube/coordinator.py index 71ca3d43fb23..3489d200f0c5 100644 --- a/homeassistant/components/youtube/coordinator.py +++ b/homeassistant/components/youtube/coordinator.py @@ -2,7 +2,7 @@ import asyncio from datetime import timedelta -from typing import Any, override +from typing import Any, Final, override from youtubeaio.types import UnauthorizedError, YouTubeBackendError @@ -33,6 +33,11 @@ from .const import ( type YouTubeConfigEntry = ConfigEntry[YouTubeDataUpdateCoordinator] +# Twice youtubeaio's 10s per-call timeout: single slow calls are handled by +# the library's own timeout (backend error / non-Short fallback); this budget +# only aborts the serial accumulation of several slow checks per channel. +_SHORTS_DETECTION_TIMEOUT: Final = 20 + def _build_video_dict(video: Any, is_short: bool) -> dict[str, Any]: """Build the video attribute dict shared by all video sensors.""" @@ -76,31 +81,42 @@ class YouTubeDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): channel_ids = self.config_entry.options[CONF_CHANNELS] try: async for channel in youtube.get_channels(channel_ids): - # Fetch up to 10 recent videos to find a Short and a non-Short. - videos = [ - v - async for v in youtube.get_playlist_items( - channel.upload_playlist_id, 10 - ) - ] - LOGGER.debug( - "Fetched %d videos for channel %s", len(videos), channel.channel_id - ) - is_short_flags = await self._get_is_short_flags(youtube, videos) - + 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 - for video, is_short in zip(videos, is_short_flags, strict=False): - 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 + 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 + ) res[channel.channel_id] = { ATTR_ID: channel.channel_id, @@ -119,27 +135,25 @@ class YouTubeDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): raise UpdateFailed("Couldn't connect to YouTube") from err return res - async def _get_is_short_flags(self, youtube: Any, videos: list[Any]) -> list[bool]: - """Return is_short flags for each video, using cache when available.""" - uncached = [ - v for v in videos if v.content_details.video_id not in self._is_short_cache - ] - if uncached: - results = await asyncio.gather( - *[youtube.is_short(v.content_details.video_id) for v in uncached], - return_exceptions=True, + async def _resolve_is_short(self, youtube: Any, video: Any) -> bool: + """Return whether a single video is a Short. + + Uses the cache when available. Videos can stop being checked as soon + as both a Short and a non-Short have been found. On error the result + is treated as non-Short without caching so the next refresh can retry. + """ + video_id = video.content_details.video_id + if video_id in self._is_short_cache: + return self._is_short_cache[video_id] + try: + result = await youtube.is_short(video_id) + except Exception as exc: # noqa: BLE001 + LOGGER.warning( + "Error determining if video %s is a Short; treating as non-Short: %s", + video_id, + exc, ) - for video, result in zip(uncached, results, strict=False): - if isinstance(result, Exception): - LOGGER.warning( - "Error determining if video %s is a Short; " - "treating as non-Short: %s", - video.content_details.video_id, - result, - ) - # Don't cache on error — let the next refresh retry. - else: - self._is_short_cache[video.content_details.video_id] = bool(result) - return [ - self._is_short_cache.get(v.content_details.video_id, False) for v in videos - ] + return False + is_short = bool(result) + self._is_short_cache[video_id] = is_short + return is_short diff --git a/tests/components/youtube/__init__.py b/tests/components/youtube/__init__.py index 65e03b44f030..f093b71cd61f 100644 --- a/tests/components/youtube/__init__.py +++ b/tests/components/youtube/__init__.py @@ -1,5 +1,6 @@ """Tests for the YouTube integration.""" +import asyncio from collections.abc import AsyncGenerator from youtubeaio.models import YouTubeChannel, YouTubePlaylistItem, YouTubeSubscription @@ -23,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, + short_check_delay: float = 0.0, ) -> None: """Initialize mock service.""" self.hass = hass @@ -30,6 +32,9 @@ 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._short_check_delay = short_check_delay + self.playlist_item_requests = 0 + self.playlist_items_yielded = 0 async def set_user_authentication( self, token: str, scopes: list[AuthScope] @@ -59,12 +64,26 @@ class MockYouTube: async def get_playlist_items( self, playlist_id: str, amount: int ) -> AsyncGenerator[YouTubePlaylistItem]: - """Get channels.""" + """Get playlist items, paginating like the real API. + + Cycles the fixture items forever in pages of `amount` (like a large + upload playlist), so tests can assert how far the coordinator + iterates before stopping. + """ channels = await async_load_json_object_fixture( self.hass, self._playlist_items_fixture, DOMAIN ) - for item in channels["items"]: - yield YouTubePlaylistItem(**item) + items = channels["items"] + if not items: + self.playlist_item_requests += 1 + return + index = 0 + while True: + self.playlist_item_requests += 1 + for _ in range(amount): + self.playlist_items_yielded += 1 + yield YouTubePlaylistItem(**items[index % len(items)]) + index += 1 async def get_user_subscriptions(self) -> AsyncGenerator[YouTubeSubscription]: """Get channels for authenticated user.""" @@ -80,4 +99,5 @@ class MockYouTube: async def is_short(self, video_id: str) -> bool: """Return whether the video is a Short.""" + await asyncio.sleep(self._short_check_delay) return video_id in self._short_video_ids diff --git a/tests/components/youtube/test_sensor.py b/tests/components/youtube/test_sensor.py index 488246149cb1..7a4dc18096c3 100644 --- a/tests/components/youtube/test_sensor.py +++ b/tests/components/youtube/test_sensor.py @@ -4,6 +4,7 @@ import asyncio from datetime import timedelta from unittest.mock import patch +import pytest from syrupy.assertion import SnapshotAssertion from youtubeaio.types import UnauthorizedError, YouTubeBackendError @@ -77,19 +78,11 @@ async def test_sensor_with_short( hass: HomeAssistant, snapshot: SnapshotAssertion, setup_integration: ComponentSetup ) -> None: """Test sensors when the channel has a Short upload.""" - await setup_integration() - with patch( "homeassistant.components.youtube.api.AsyncConfigEntryAuth.get_resource", return_value=MockYouTube(hass, short_video_ids={"wysukDrMdqU"}), ): - # Clear the coordinator's is_short cache so the Short is re-detected. - entry = hass.config_entries.async_entries(DOMAIN)[0] - entry.runtime_data._is_short_cache.clear() - future = dt_util.utcnow() + timedelta(minutes=15) - async_fire_time_changed(hass, future) - await hass.async_block_till_done() - await asyncio.sleep(0.1) + await setup_integration() state = hass.states.get("sensor.google_for_developers_latest_short") assert state == snapshot @@ -202,3 +195,86 @@ async def test_sensor_unavailable( state = hass.states.get("sensor.google_for_developers_views") assert state.state == "unavailable" + + +@pytest.mark.parametrize( + ("short_video_ids", "expected_yielded", "expected_latest_video_state"), + [ + pytest.param( + {"wysukDrMdqU"}, + 2, + "Google I/O 2023 Developer Keynote in 5 minutes", + id="stops_after_short_and_non_short_found", + ), + pytest.param( + {"wysukDrMdqU", "hleLlcHwQLM", "lMKjtSFujcw", "c0mqBuXPrpA", "_n9xwuTORas"}, + 10, + "unavailable", + id="caps_at_page_size_when_only_shorts", + ), + ], +) +async def test_sensor_playlist_iteration( + hass: HomeAssistant, + setup_integration: ComponentSetup, + short_video_ids: set[str], + expected_yielded: int, + expected_latest_video_state: str, +) -> None: + """Test playlist iteration stops as early as possible and never paginates.""" + mock = MockYouTube(hass, short_video_ids=short_video_ids) + with patch( + "homeassistant.components.youtube.api.AsyncConfigEntryAuth.get_resource", + return_value=mock, + ): + await setup_integration() + + assert mock.playlist_item_requests == 1 + assert mock.playlist_items_yielded == expected_yielded + + state = hass.states.get("sensor.google_for_developers_latest_short") + assert state.state == "What's new in Google Home in less than 1 minute" + + state = hass.states.get("sensor.google_for_developers_latest_video") + assert state.state == expected_latest_video_state + + +async def test_sensor_shorts_detection_timeout( + hass: HomeAssistant, + caplog: pytest.LogCaptureFixture, + setup_integration: ComponentSetup, +) -> None: + """Test the per-channel pass is bounded when is_short hangs. + + Channel-level sensors stay available while the video sensors report + unavailable from the partial results. + """ + mock = MockYouTube(hass, short_check_delay=1) + with ( + patch( + "homeassistant.components.youtube.api.AsyncConfigEntryAuth.get_resource", + return_value=mock, + ), + patch( + "homeassistant.components.youtube.coordinator._SHORTS_DETECTION_TIMEOUT", + 0.1, + ), + ): + await setup_integration() + + assert "Timed out processing recent uploads" in caplog.text + + state = hass.states.get("sensor.google_for_developers_latest_upload") + assert state.state == "unavailable" + + state = hass.states.get("sensor.google_for_developers_latest_short") + assert state.state == "unavailable" + + state = hass.states.get("sensor.google_for_developers_latest_video") + assert state.state == "unavailable" + + state = hass.states.get("sensor.google_for_developers_subscribers") + assert state.state == "2290000" + + state = hass.states.get("sensor.google_for_developers_views") + assert state.state == "214141263"