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

Bump TwitchAPI to 3.10.0 (#92418)

Browse source 
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
This commit is contained in:
Joost Lekkerkerker 2023-05-21 21:13:59 +02:00 • committed by GitHub
parent e27554f7a6
commit c12fae4775
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 338 additions and 148 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
homeassistant/components/twitch/manifest.json
View file

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

106
homeassistant/components/twitch/sensor.py
View file

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

2
requirements_all.txt
View file

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

2
requirements_test_all.txt
View file

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

189
tests/components/twitch/__init__.py
View file

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

185
tests/components/twitch/test_twitch.py
View file

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