From c12fae47751e7f95fd17aeb0d1fefeddcd407ff0 Mon Sep 17 00:00:00 2001 From: Joost Lekkerkerker Date: Sun, 21 May 2023 21:13:59 +0200 Subject: [PATCH] Bump TwitchAPI to 3.10.0 (#92418) Co-authored-by: Franck Nijhof --- homeassistant/components/twitch/manifest.json | 2 +- homeassistant/components/twitch/sensor.py | 106 +++++----- requirements_all.txt | 2 +- requirements_test_all.txt | 2 +- tests/components/twitch/__init__.py | 189 ++++++++++++++++++ tests/components/twitch/test_twitch.py | 185 +++++++++-------- 6 files changed, 338 insertions(+), 148 deletions(-) diff --git a/homeassistant/components/twitch/manifest.json b/homeassistant/components/twitch/manifest.json index b954db7270be..c11be26c45a9 100644 --- a/homeassistant/components/twitch/manifest.json +++ b/homeassistant/components/twitch/manifest.json @@ -5,5 +5,5 @@ "documentation": "https://www.home-assistant.io/integrations/twitch", "iot_class": "cloud_polling", "loggers": ["twitch"], - "requirements": ["twitchAPI==2.5.2"] + "requirements": ["twitchAPI==3.10.0"] } diff --git a/homeassistant/components/twitch/sensor.py b/homeassistant/components/twitch/sensor.py index 63746aae46fb..fcc8225b73c9 100644 --- a/homeassistant/components/twitch/sensor.py +++ b/homeassistant/components/twitch/sensor.py @@ -3,13 +3,17 @@ from __future__ import annotations import logging +from twitchAPI.helper import first from twitchAPI.twitch import ( AuthScope, AuthType, InvalidTokenException, MissingScopeException, Twitch, + TwitchAPIException, TwitchAuthorizationException, + TwitchResourceNotFound, + TwitchUser, ) import voluptuous as vol @@ -51,10 +55,10 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( ) -def setup_platform( +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 Twitch platform.""" @@ -64,7 +68,7 @@ def setup_platform( oauth_token = config.get(CONF_TOKEN) try: - client = Twitch( + client = await Twitch( app_id=client_id, app_secret=client_secret, target_app_auth_scope=OAUTH_SCOPES, @@ -76,7 +80,7 @@ def setup_platform( if oauth_token: try: - client.set_user_authentication( + await client.set_user_authentication( token=oauth_token, scope=OAUTH_SCOPES, validate=True ) except MissingScopeException: @@ -86,71 +90,45 @@ def setup_platform( _LOGGER.error("OAuth token is invalid") return - channels = client.get_users(logins=channels) + twitch_users: list[TwitchUser] = [] + async for channel in client.get_users(logins=channels): + twitch_users.append(channel) - add_entities( - [TwitchSensor(channel, client) for channel in channels["data"]], + async_add_entities( + [TwitchSensor(channel, client) for channel in twitch_users], True, ) class TwitchSensor(SensorEntity): - """Representation of an Twitch channel.""" + """Representation of a Twitch channel.""" _attr_icon = ICON - def __init__(self, channel: dict[str, str], client: Twitch) -> None: + def __init__(self, channel: TwitchUser, client: Twitch) -> None: """Initialize the sensor.""" self._client = client + self._channel = channel self._enable_user_auth = client.has_required_auth(AuthType.USER, OAUTH_SCOPES) - self._attr_name = channel["display_name"] - self._attr_unique_id = channel["id"] + self._attr_name = channel.display_name + self._attr_unique_id = channel.id - def update(self) -> None: + async def async_update(self) -> None: """Update device state.""" - followers = self._client.get_users_follows(to_id=self.unique_id)["total"] - channel = self._client.get_users(user_ids=[self.unique_id])["data"][0] + followers = (await self._client.get_users_follows(to_id=self._channel.id)).total self._attr_extra_state_attributes = { ATTR_FOLLOWING: followers, - ATTR_VIEWS: channel["view_count"], + ATTR_VIEWS: self._channel.view_count, } if self._enable_user_auth: - user = self._client.get_users()["data"][0]["id"] - - subs = self._client.check_user_subscription( - user_id=user, broadcaster_id=self.unique_id - ) - if "data" in subs: - self._attr_extra_state_attributes[ATTR_SUBSCRIPTION] = True - self._attr_extra_state_attributes[ATTR_SUBSCRIPTION_GIFTED] = subs[ - "data" - ][0]["is_gift"] - elif "status" in subs and subs["status"] == 404: - self._attr_extra_state_attributes[ATTR_SUBSCRIPTION] = False - elif "error" in subs: - _LOGGER.error( - "Error response on check_user_subscription: %s", subs["error"] - ) - return - else: - _LOGGER.error("Unknown error response on check_user_subscription") - return - - follows = self._client.get_users_follows( - from_id=user, to_id=self.unique_id - )["data"] - self._attr_extra_state_attributes[ATTR_FOLLOW] = len(follows) > 0 - if len(follows): - self._attr_extra_state_attributes[ATTR_FOLLOW_SINCE] = follows[0][ - "followed_at" - ] - - if streams := self._client.get_streams(user_id=[self.unique_id])["data"]: - stream = streams[0] + await self._async_add_user_attributes() + if stream := ( + await first(self._client.get_streams(user_id=[self._channel.id], first=1)) + ): self._attr_native_value = STATE_STREAMING - self._attr_extra_state_attributes[ATTR_GAME] = stream["game_name"] - self._attr_extra_state_attributes[ATTR_TITLE] = stream["title"] - self._attr_entity_picture = stream["thumbnail_url"] + self._attr_extra_state_attributes[ATTR_GAME] = stream.game_name + self._attr_extra_state_attributes[ATTR_TITLE] = stream.title + self._attr_entity_picture = stream.thumbnail_url if self._attr_entity_picture is not None: self._attr_entity_picture = self._attr_entity_picture.format( height=24, @@ -160,4 +138,30 @@ class TwitchSensor(SensorEntity): self._attr_native_value = STATE_OFFLINE self._attr_extra_state_attributes[ATTR_GAME] = None self._attr_extra_state_attributes[ATTR_TITLE] = None - self._attr_entity_picture = channel["profile_image_url"] + self._attr_entity_picture = self._channel.profile_image_url + + async def _async_add_user_attributes(self) -> None: + if not (user := await first(self._client.get_users())): + return + self._attr_extra_state_attributes[ATTR_SUBSCRIPTION] = False + try: + sub = await self._client.check_user_subscription( + user_id=user.id, broadcaster_id=self._channel.id + ) + self._attr_extra_state_attributes[ATTR_SUBSCRIPTION] = True + self._attr_extra_state_attributes[ATTR_SUBSCRIPTION_GIFTED] = sub.is_gift + except TwitchResourceNotFound: + _LOGGER.debug("User is not subscribed") + except TwitchAPIException as exc: + _LOGGER.error("Error response on check_user_subscription: %s", exc) + + follows = ( + await self._client.get_users_follows( + from_id=user.id, to_id=self._channel.id + ) + ).data + self._attr_extra_state_attributes[ATTR_FOLLOW] = len(follows) > 0 + if len(follows): + self._attr_extra_state_attributes[ATTR_FOLLOW_SINCE] = follows[ + 0 + ].followed_at diff --git a/requirements_all.txt b/requirements_all.txt index 4094cc2acc56..965ff04fd0f3 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -2556,7 +2556,7 @@ twentemilieu==1.0.0 twilio==6.32.0 # homeassistant.components.twitch -twitchAPI==2.5.2 +twitchAPI==3.10.0 # homeassistant.components.ukraine_alarm uasiren==0.0.1 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 05954c0f9e6a..cb197823f1d4 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1847,7 +1847,7 @@ twentemilieu==1.0.0 twilio==6.32.0 # homeassistant.components.twitch -twitchAPI==2.5.2 +twitchAPI==3.10.0 # homeassistant.components.ukraine_alarm uasiren==0.0.1 diff --git a/tests/components/twitch/__init__.py b/tests/components/twitch/__init__.py index ec26cf264ef7..5c371a0e2ee4 100644 --- a/tests/components/twitch/__init__.py +++ b/tests/components/twitch/__init__.py @@ -1 +1,190 @@ """Tests for the Twitch component.""" +import asyncio +from collections.abc import AsyncGenerator +from dataclasses import dataclass +from typing import Any, Optional + +from twitchAPI.object import TwitchUser +from twitchAPI.twitch import ( + InvalidTokenException, + MissingScopeException, + TwitchAPIException, + TwitchAuthorizationException, + TwitchResourceNotFound, +) +from twitchAPI.types import AuthScope, AuthType + +USER_OBJECT: TwitchUser = TwitchUser( + id=123, + display_name="channel123", + offline_image_url="logo.png", + profile_image_url="logo.png", + view_count=42, +) + + +class TwitchUserFollowResultMock: + """Mock for twitch user follow result.""" + + def __init__(self, follows: list[dict[str, Any]]) -> None: + """Initialize mock.""" + self.total = len(follows) + self.data = follows + + +@dataclass +class UserSubscriptionMock: + """User subscription mock.""" + + broadcaster_id: str + is_gift: bool + + +@dataclass +class UserFollowMock: + """User follow mock.""" + + followed_at: str + + +@dataclass +class StreamMock: + """Stream mock.""" + + game_name: str + title: str + thumbnail_url: str + + +STREAMS = StreamMock( + game_name="Good game", title="Title", thumbnail_url="stream-medium.png" +) + + +class TwitchMock: + """Mock for the twitch object.""" + + def __await__(self): + """Add async capabilities to the mock.""" + t = asyncio.create_task(self._noop()) + yield from t + return self + + def __init__( + self, + is_streaming: bool = True, + is_gifted: bool = False, + is_subscribed: bool = False, + is_following: bool = True, + ) -> None: + """Initialize mock.""" + self._is_streaming = is_streaming + self._is_gifted = is_gifted + self._is_subscribed = is_subscribed + self._is_following = is_following + + async def _noop(self): + """Fake function to create task.""" + pass + + async def get_users( + self, user_ids: Optional[list[str]] = None, logins: Optional[list[str]] = None + ) -> AsyncGenerator[TwitchUser, None]: + """Get list of mock users.""" + for user in [USER_OBJECT]: + yield user + + def has_required_auth( + self, required_type: AuthType, required_scope: list[AuthScope] + ) -> bool: + """Return if auth required.""" + return True + + async def get_users_follows( + self, to_id: Optional[str] = None, from_id: Optional[str] = None + ) -> TwitchUserFollowResultMock: + """Return the followers of the user.""" + if self._is_following: + return TwitchUserFollowResultMock( + follows=[UserFollowMock("2020-01-20T21:22:42") for _ in range(0, 24)] + ) + return TwitchUserFollowResultMock(follows=[]) + + async def check_user_subscription( + self, broadcaster_id: str, user_id: str + ) -> UserSubscriptionMock: + """Check if the user is subscribed.""" + if self._is_subscribed: + return UserSubscriptionMock( + broadcaster_id=broadcaster_id, is_gift=self._is_gifted + ) + raise TwitchResourceNotFound + + async def set_user_authentication( + self, token: str, scope: list[AuthScope], validate: bool = True + ) -> None: + """Set user authentication.""" + pass + + async def get_streams( + self, user_id: list[str], first: int + ) -> AsyncGenerator[StreamMock, None]: + """Get streams for the user.""" + streams = [] + if self._is_streaming: + streams = [STREAMS] + for stream in streams: + yield stream + + +class TwitchUnauthorizedMock(TwitchMock): + """Twitch mock to test if the client is unauthorized.""" + + def __await__(self): + """Add async capabilities to the mock.""" + raise TwitchAuthorizationException() + + +class TwitchMissingScopeMock(TwitchMock): + """Twitch mock to test missing scopes.""" + + async def set_user_authentication( + self, token: str, scope: list[AuthScope], validate: bool = True + ) -> None: + """Set user authentication.""" + raise MissingScopeException() + + +class TwitchInvalidTokenMock(TwitchMock): + """Twitch mock to test invalid token.""" + + async def set_user_authentication( + self, token: str, scope: list[AuthScope], validate: bool = True + ) -> None: + """Set user authentication.""" + raise InvalidTokenException() + + +class TwitchInvalidUserMock(TwitchMock): + """Twitch mock to test invalid user.""" + + async def get_users( + self, user_ids: Optional[list[str]] = None, logins: Optional[list[str]] = None + ) -> AsyncGenerator[TwitchUser, None]: + """Get list of mock users.""" + if user_ids is not None or logins is not None: + async for user in super().get_users(user_ids, logins): + yield user + else: + for user in []: + yield user + + +class TwitchAPIExceptionMock(TwitchMock): + """Twitch mock to test when twitch api throws unknown exception.""" + + async def check_user_subscription( + self, broadcaster_id: str, user_id: str + ) -> UserSubscriptionMock: + """Check if the user is subscribed.""" + raise TwitchAPIException() diff --git a/tests/components/twitch/test_twitch.py b/tests/components/twitch/test_twitch.py index fb932c8f35cc..4a33831dd32e 100644 --- a/tests/components/twitch/test_twitch.py +++ b/tests/components/twitch/test_twitch.py @@ -1,11 +1,20 @@ """The tests for an update of the Twitch component.""" -from unittest.mock import MagicMock, patch +from unittest.mock import patch from homeassistant.components import sensor from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET from homeassistant.core import HomeAssistant from homeassistant.setup import async_setup_component +from . import ( + TwitchAPIExceptionMock, + TwitchInvalidTokenMock, + TwitchInvalidUserMock, + TwitchMissingScopeMock, + TwitchMock, + TwitchUnauthorizedMock, +) + ENTITY_ID = "sensor.channel123" CONFIG = { sensor.DOMAIN: { @@ -25,41 +34,13 @@ CONFIG_WITH_OAUTH = { } } -USER_OBJECT = { - "id": 123, - "display_name": "channel123", - "offline_image_url": "logo.png", - "profile_image_url": "logo.png", - "view_count": 42, -} -STREAM_OBJECT_ONLINE = { - "game_name": "Good Game", - "title": "Title", - "thumbnail_url": "stream-medium.png", -} - -FOLLOWERS_OBJECT = [{"followed_at": "2020-01-20T21:22:42"}] * 24 -OAUTH_USER_ID = {"id": 987} -SUB_ACTIVE = {"is_gift": False} -FOLLOW_ACTIVE = {"followed_at": "2020-01-20T21:22:42"} - - -def make_data(data): - """Create a data object.""" - return {"data": data, "total": len(data)} - async def test_init(hass: HomeAssistant) -> None: """Test initial config.""" - twitch_mock = MagicMock() - twitch_mock.get_streams.return_value = make_data([]) - twitch_mock.get_users.return_value = make_data([USER_OBJECT]) - twitch_mock.get_users_follows.return_value = make_data(FOLLOWERS_OBJECT) - twitch_mock.has_required_auth.return_value = False - with patch( - "homeassistant.components.twitch.sensor.Twitch", return_value=twitch_mock + "homeassistant.components.twitch.sensor.Twitch", + return_value=TwitchMock(is_streaming=False), ): assert await async_setup_component(hass, sensor.DOMAIN, CONFIG) is True await hass.async_block_till_done() @@ -76,15 +57,9 @@ async def test_init(hass: HomeAssistant) -> None: async def test_offline(hass: HomeAssistant) -> None: """Test offline state.""" - twitch_mock = MagicMock() - twitch_mock.get_streams.return_value = make_data([]) - twitch_mock.get_users.return_value = make_data([USER_OBJECT]) - twitch_mock.get_users_follows.return_value = make_data(FOLLOWERS_OBJECT) - twitch_mock.has_required_auth.return_value = False - with patch( "homeassistant.components.twitch.sensor.Twitch", - return_value=twitch_mock, + return_value=TwitchMock(is_streaming=False), ): assert await async_setup_component(hass, sensor.DOMAIN, CONFIG) is True await hass.async_block_till_done() @@ -97,15 +72,9 @@ async def test_offline(hass: HomeAssistant) -> None: async def test_streaming(hass: HomeAssistant) -> None: """Test streaming state.""" - twitch_mock = MagicMock() - twitch_mock.get_users.return_value = make_data([USER_OBJECT]) - twitch_mock.get_users_follows.return_value = make_data(FOLLOWERS_OBJECT) - twitch_mock.get_streams.return_value = make_data([STREAM_OBJECT_ONLINE]) - twitch_mock.has_required_auth.return_value = False - with patch( "homeassistant.components.twitch.sensor.Twitch", - return_value=twitch_mock, + return_value=TwitchMock(), ): assert await async_setup_component(hass, sensor.DOMAIN, CONFIG) is True await hass.async_block_till_done() @@ -113,30 +82,16 @@ async def test_streaming(hass: HomeAssistant) -> None: sensor_state = hass.states.get(ENTITY_ID) assert sensor_state.state == "streaming" assert sensor_state.attributes["entity_picture"] == "stream-medium.png" - assert sensor_state.attributes["game"] == "Good Game" + assert sensor_state.attributes["game"] == "Good game" assert sensor_state.attributes["title"] == "Title" async def test_oauth_without_sub_and_follow(hass: HomeAssistant) -> None: """Test state with oauth.""" - twitch_mock = MagicMock() - twitch_mock.get_streams.return_value = make_data([]) - twitch_mock.get_users.side_effect = [ - make_data([USER_OBJECT]), - make_data([USER_OBJECT]), - make_data([OAUTH_USER_ID]), - ] - twitch_mock.get_users_follows.side_effect = [ - make_data(FOLLOWERS_OBJECT), - make_data([]), - ] - twitch_mock.has_required_auth.return_value = True - twitch_mock.check_user_subscription.return_value = {"status": 404} - with patch( "homeassistant.components.twitch.sensor.Twitch", - return_value=twitch_mock, + return_value=TwitchMock(is_following=False), ): assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH) await hass.async_block_till_done() @@ -149,25 +104,11 @@ async def test_oauth_without_sub_and_follow(hass: HomeAssistant) -> None: async def test_oauth_with_sub(hass: HomeAssistant) -> None: """Test state with oauth and sub.""" - twitch_mock = MagicMock() - twitch_mock.get_streams.return_value = make_data([]) - twitch_mock.get_users.side_effect = [ - make_data([USER_OBJECT]), - make_data([USER_OBJECT]), - make_data([OAUTH_USER_ID]), - ] - twitch_mock.get_users_follows.side_effect = [ - make_data(FOLLOWERS_OBJECT), - make_data([]), - ] - twitch_mock.has_required_auth.return_value = True - - # This function does not return an array so use make_data - twitch_mock.check_user_subscription.return_value = make_data([SUB_ACTIVE]) - with patch( "homeassistant.components.twitch.sensor.Twitch", - return_value=twitch_mock, + return_value=TwitchMock( + is_subscribed=True, is_gifted=False, is_following=False + ), ): assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH) await hass.async_block_till_done() @@ -181,28 +122,84 @@ async def test_oauth_with_sub(hass: HomeAssistant) -> None: async def test_oauth_with_follow(hass: HomeAssistant) -> None: """Test state with oauth and follow.""" - twitch_mock = MagicMock() - twitch_mock.get_streams.return_value = make_data([]) - twitch_mock.get_users.side_effect = [ - make_data([USER_OBJECT]), - make_data([USER_OBJECT]), - make_data([OAUTH_USER_ID]), - ] - twitch_mock.get_users_follows.side_effect = [ - make_data(FOLLOWERS_OBJECT), - make_data([FOLLOW_ACTIVE]), - ] - twitch_mock.has_required_auth.return_value = True - twitch_mock.check_user_subscription.return_value = {"status": 404} + with patch( + "homeassistant.components.twitch.sensor.Twitch", + return_value=TwitchMock(), + ): + assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH) + await hass.async_block_till_done() + + sensor_state = hass.states.get(ENTITY_ID) + assert sensor_state.attributes["following"] is True + assert sensor_state.attributes["following_since"] == "2020-01-20T21:22:42" + + +async def test_auth_with_invalid_credentials(hass: HomeAssistant) -> None: + """Test auth with invalid credentials.""" with patch( "homeassistant.components.twitch.sensor.Twitch", - return_value=twitch_mock, + return_value=TwitchUnauthorizedMock(), + ): + assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH) + await hass.async_block_till_done() + + sensor_state = hass.states.get(ENTITY_ID) + assert sensor_state is None + + +async def test_auth_with_missing_scope(hass: HomeAssistant) -> None: + """Test auth with invalid credentials.""" + + with patch( + "homeassistant.components.twitch.sensor.Twitch", + return_value=TwitchMissingScopeMock(), + ): + assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH) + await hass.async_block_till_done() + + sensor_state = hass.states.get(ENTITY_ID) + assert sensor_state is None + + +async def test_auth_with_invalid_token(hass: HomeAssistant) -> None: + """Test auth with invalid credentials.""" + + with patch( + "homeassistant.components.twitch.sensor.Twitch", + return_value=TwitchInvalidTokenMock(), + ): + assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH) + await hass.async_block_till_done() + + sensor_state = hass.states.get(ENTITY_ID) + assert sensor_state is None + + +async def test_auth_with_invalid_user(hass: HomeAssistant) -> None: + """Test auth with invalid user.""" + + with patch( + "homeassistant.components.twitch.sensor.Twitch", + return_value=TwitchInvalidUserMock(), + ): + assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH) + await hass.async_block_till_done() + + sensor_state = hass.states.get(ENTITY_ID) + assert "subscribed" not in sensor_state.attributes + + +async def test_auth_with_api_exception(hass: HomeAssistant) -> None: + """Test auth with invalid user.""" + + with patch( + "homeassistant.components.twitch.sensor.Twitch", + return_value=TwitchAPIExceptionMock(), ): assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH) await hass.async_block_till_done() sensor_state = hass.states.get(ENTITY_ID) assert sensor_state.attributes["subscribed"] is False - assert sensor_state.attributes["following"] is True - assert sensor_state.attributes["following_since"] == "2020-01-20T21:22:42" + assert "subscription_is_gifted" not in sensor_state.attributes