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

Fix Twitch component to use new API (#67153)

Browse source 
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
Mark Dietzer 2022-02-25 09:15:49 -08:00 • committed by GitHub
parent a4ba71408b
commit 07a792019e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 164 additions and 102 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
homeassistant/components/twitch/__init__.py
View file

@ -1 +1 @@
"""The twitch component."""
"""The Twitch component."""

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

@ -2,7 +2,7 @@
"domain": "twitch",
"name": "Twitch",
"documentation": "https://www.home-assistant.io/integrations/twitch",
"requirements": ["python-twitch-client==0.6.0"],
"requirements": ["twitchAPI==2.5.2"],
"codeowners": [],
"iot_class": "cloud_polling",
"loggers": ["twitch"]

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

@ -3,12 +3,18 @@ from __future__ import annotations
import logging
from requests.exceptions import HTTPError
from twitch import TwitchClient
from twitchAPI.twitch import (
AuthScope,
AuthType,
InvalidTokenException,
MissingScopeException,
Twitch,
TwitchAuthorizationException,
)
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
from homeassistant.const import CONF_CLIENT_ID, CONF_TOKEN
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET, CONF_TOKEN
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -33,9 +39,12 @@ ICON = "mdi:twitch"
STATE_OFFLINE = "offline"
STATE_STREAMING = "streaming"
OAUTH_SCOPES = [AuthScope.USER_READ_SUBSCRIPTIONS]
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_CLIENT_ID): cv.string,
vol.Required(CONF_CLIENT_SECRET): cv.string,
vol.Required(CONF_CHANNELS): vol.All(cv.ensure_list, [cv.string]),
vol.Optional(CONF_TOKEN): cv.string,
}
@ -51,28 +60,45 @@ def setup_platform(
"""Set up the Twitch platform."""
channels = config[CONF_CHANNELS]
client_id = config[CONF_CLIENT_ID]
client_secret = config[CONF_CLIENT_SECRET]
oauth_token = config.get(CONF_TOKEN)
client = TwitchClient(client_id, oauth_token)
client = Twitch(app_id=client_id, app_secret=client_secret)
client.auto_refresh_auth = False
try:
client.ingests.get_server_list()
except HTTPError:
_LOGGER.error("Client ID or OAuth token is not valid")
client.authenticate_app(scope=OAUTH_SCOPES)
except TwitchAuthorizationException:
_LOGGER.error("INvalid client ID or client secret")
return
channel_ids = client.users.translate_usernames_to_ids(channels)
if oauth_token:
try:
client.set_user_authentication(
token=oauth_token, scope=OAUTH_SCOPES, validate=True
)
except MissingScopeException:
_LOGGER.error("OAuth token is missing required scope")
return
except InvalidTokenException:
_LOGGER.error("OAuth token is invalid")
return
add_entities([TwitchSensor(channel_id, client) for channel_id in channel_ids], True)
channels = client.get_users(logins=channels)
add_entities(
[TwitchSensor(channel=channel, client=client) for channel in channels["data"]],
True,
)
class TwitchSensor(SensorEntity):
"""Representation of an Twitch channel."""
def __init__(self, channel, client):
def __init__(self, channel, client: Twitch):
"""Initialize the sensor."""
self._client = client
self._channel = channel
self._oauth_enabled = client._oauth_token is not None
self._enable_user_auth = client.has_required_auth(AuthType.USER, OAUTH_SCOPES)
self._state = None
self._preview = None
self._game = None
@ -84,7 +110,7 @@ class TwitchSensor(SensorEntity):
@property
def name(self):
"""Return the name of the sensor."""
return self._channel.display_name
return self._channel["display_name"]
@property
def native_value(self):
@ -101,7 +127,7 @@ class TwitchSensor(SensorEntity):
"""Return the state attributes."""
attr = dict(self._statistics)
if self._oauth_enabled:
if self._enable_user_auth:
attr.update(self._subscription)
attr.update(self._follow)
@ -112,7 +138,7 @@ class TwitchSensor(SensorEntity):
@property
def unique_id(self):
"""Return unique ID for this sensor."""
return self._channel.id
return self._channel["id"]
@property
def icon(self):
@ -122,41 +148,51 @@ class TwitchSensor(SensorEntity):
def update(self):
"""Update device state."""
channel = self._client.channels.get_by_id(self._channel.id)
followers = self._client.get_users_follows(to_id=self._channel["id"])["total"]
channel = self._client.get_users(user_ids=[self._channel["id"]])["data"][0]
self._statistics = {
ATTR_FOLLOWING: channel.followers,
ATTR_VIEWS: channel.views,
ATTR_FOLLOWING: followers,
ATTR_VIEWS: channel["view_count"],
}
if self._oauth_enabled:
user = self._client.users.get()
if self._enable_user_auth:
user = self._client.get_users()["data"][0]
try:
sub = self._client.users.check_subscribed_to_channel(
user.id, self._channel.id
subs = self._client.check_user_subscription(
user_id=user["id"], broadcaster_id=self._channel["id"]
)
if "data" in subs:
self._subscription = {
ATTR_SUBSCRIPTION: True,
ATTR_SUBSCRIPTION_SINCE: sub.created_at,
ATTR_SUBSCRIPTION_GIFTED: sub.is_gift,
ATTR_SUBSCRIPTION_GIFTED: subs["data"][0]["is_gift"],
}
except HTTPError:
elif "status" in subs and subs["status"] == 404:
self._subscription = {ATTR_SUBSCRIPTION: False}
try:
follow = self._client.users.check_follows_channel(
user.id, self._channel.id
elif "error" in subs:
raise Exception(
f"Error response on check_user_subscription: {subs['error']}"
)
self._follow = {ATTR_FOLLOW: True, ATTR_FOLLOW_SINCE: follow.created_at}
except HTTPError:
else:
raise Exception("Unknown error response on check_user_subscription")
follows = self._client.get_users_follows(
from_id=user["id"], to_id=self._channel["id"]
)["data"]
if len(follows) > 0:
self._follow = {
ATTR_FOLLOW: True,
ATTR_FOLLOW_SINCE: follows[0]["followed_at"],
}
else:
self._follow = {ATTR_FOLLOW: False}
stream = self._client.streams.get_stream_by_user(self._channel.id)
if stream:
self._game = stream.channel.get("game")
self._title = stream.channel.get("status")
self._preview = stream.preview.get("medium")
streams = self._client.get_streams(user_id=[self._channel["id"]])["data"]
if len(streams) > 0:
stream = streams[0]
self._game = stream["game_name"]
self._title = stream["title"]
self._preview = stream["thumbnail_url"]
self._state = STATE_STREAMING
else:
self._preview = self._channel.logo
self._preview = channel["offline_image_url"]
self._state = STATE_OFFLINE

6
requirements_all.txt
View file

@ -1954,9 +1954,6 @@ python-tado==0.12.0
# homeassistant.components.telegram_bot
python-telegram-bot==13.1
# homeassistant.components.twitch
python-twitch-client==0.6.0
# homeassistant.components.vlc
python-vlc==1.1.2
@ -2360,6 +2357,9 @@ twentemilieu==0.5.0
# homeassistant.components.twilio
twilio==6.32.0
# homeassistant.components.twitch
twitchAPI==2.5.2
# homeassistant.components.rainforest_eagle
uEagle==0.0.2

6
requirements_test_all.txt
View file

@ -1235,9 +1235,6 @@ python-songpal==0.14
# homeassistant.components.tado
python-tado==0.12.0
# homeassistant.components.twitch
python-twitch-client==0.6.0
# homeassistant.components.awair
python_awair==0.2.1
@ -1467,6 +1464,9 @@ twentemilieu==0.5.0
# homeassistant.components.twilio
twilio==6.32.0
# homeassistant.components.twitch
twitchAPI==2.5.2
# homeassistant.components.rainforest_eagle
uEagle==0.0.2

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

@ -1,11 +1,8 @@
"""The tests for an update of the Twitch component."""
from unittest.mock import MagicMock, patch
from requests import HTTPError
from twitch.resources import Channel, Follow, Stream, Subscription, User
from homeassistant.components import sensor
from homeassistant.const import CONF_CLIENT_ID
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
from homeassistant.setup import async_setup_component
ENTITY_ID = "sensor.channel123"
@ -13,6 +10,7 @@ CONFIG = {
sensor.DOMAIN: {
"platform": "twitch",
CONF_CLIENT_ID: "1234",
CONF_CLIENT_SECRET: " abcd",
"channels": ["channel123"],
}
}
@ -20,39 +18,46 @@ CONFIG_WITH_OAUTH = {
sensor.DOMAIN: {
"platform": "twitch",
CONF_CLIENT_ID: "1234",
CONF_CLIENT_SECRET: "abcd",
"channels": ["channel123"],
"token": "9876",
}
}
USER_ID = User({"id": 123, "display_name": "channel123", "logo": "logo.png"})
STREAM_OBJECT_ONLINE = Stream(
{
"channel": {"game": "Good Game", "status": "Title"},
"preview": {"medium": "stream-medium.png"},
}
)
CHANNEL_OBJECT = Channel({"followers": 42, "views": 24})
OAUTH_USER_ID = User({"id": 987})
SUB_ACTIVE = Subscription({"created_at": "2020-01-20T21:22:42", "is_gift": False})
FOLLOW_ACTIVE = Follow({"created_at": "2020-01-20T21:22:42"})
USER_OBJECT = {
"id": 123,
"display_name": "channel123",
"offline_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):
"""Test initial config."""
channels = MagicMock()
channels.get_by_id.return_value = CHANNEL_OBJECT
streams = MagicMock()
streams.get_stream_by_user.return_value = None
twitch_mock = MagicMock()
twitch_mock.users.translate_usernames_to_ids.return_value = [USER_ID]
twitch_mock.channels = channels
twitch_mock.streams = streams
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.TwitchClient", return_value=twitch_mock
"homeassistant.components.twitch.sensor.Twitch", return_value=twitch_mock
):
assert await async_setup_component(hass, sensor.DOMAIN, CONFIG) is True
await hass.async_block_till_done()
@ -62,20 +67,21 @@ async def test_init(hass):
assert sensor_state.name == "channel123"
assert sensor_state.attributes["icon"] == "mdi:twitch"
assert sensor_state.attributes["friendly_name"] == "channel123"
assert sensor_state.attributes["views"] == 24
assert sensor_state.attributes["followers"] == 42
assert sensor_state.attributes["views"] == 42
assert sensor_state.attributes["followers"] == 24
async def test_offline(hass):
"""Test offline state."""
twitch_mock = MagicMock()
twitch_mock.users.translate_usernames_to_ids.return_value = [USER_ID]
twitch_mock.channels.get_by_id.return_value = CHANNEL_OBJECT
twitch_mock.streams.get_stream_by_user.return_value = None
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.TwitchClient",
"homeassistant.components.twitch.sensor.Twitch",
return_value=twitch_mock,
):
assert await async_setup_component(hass, sensor.DOMAIN, CONFIG) is True
@ -90,12 +96,13 @@ async def test_streaming(hass):
"""Test streaming state."""
twitch_mock = MagicMock()
twitch_mock.users.translate_usernames_to_ids.return_value = [USER_ID]
twitch_mock.channels.get_by_id.return_value = CHANNEL_OBJECT
twitch_mock.streams.get_stream_by_user.return_value = STREAM_OBJECT_ONLINE
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.TwitchClient",
"homeassistant.components.twitch.sensor.Twitch",
return_value=twitch_mock,
):
assert await async_setup_component(hass, sensor.DOMAIN, CONFIG) is True
@ -112,15 +119,21 @@ async def test_oauth_without_sub_and_follow(hass):
"""Test state with oauth."""
twitch_mock = MagicMock()
twitch_mock.users.translate_usernames_to_ids.return_value = [USER_ID]
twitch_mock.channels.get_by_id.return_value = CHANNEL_OBJECT
twitch_mock._oauth_token = True # A replacement for the token
twitch_mock.users.get.return_value = OAUTH_USER_ID
twitch_mock.users.check_subscribed_to_channel.side_effect = HTTPError()
twitch_mock.users.check_follows_channel.side_effect = HTTPError()
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.TwitchClient",
"homeassistant.components.twitch.sensor.Twitch",
return_value=twitch_mock,
):
assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH)
@ -135,15 +148,23 @@ async def test_oauth_with_sub(hass):
"""Test state with oauth and sub."""
twitch_mock = MagicMock()
twitch_mock.users.translate_usernames_to_ids.return_value = [USER_ID]
twitch_mock.channels.get_by_id.return_value = CHANNEL_OBJECT
twitch_mock._oauth_token = True # A replacement for the token
twitch_mock.users.get.return_value = OAUTH_USER_ID
twitch_mock.users.check_subscribed_to_channel.return_value = SUB_ACTIVE
twitch_mock.users.check_follows_channel.side_effect = HTTPError()
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.TwitchClient",
"homeassistant.components.twitch.sensor.Twitch",
return_value=twitch_mock,
):
assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH)
@ -151,7 +172,6 @@ async def test_oauth_with_sub(hass):
sensor_state = hass.states.get(ENTITY_ID)
assert sensor_state.attributes["subscribed"] is True
assert sensor_state.attributes["subscribed_since"] == "2020-01-20T21:22:42"
assert sensor_state.attributes["subscription_is_gifted"] is False
assert sensor_state.attributes["following"] is False
@ -160,15 +180,21 @@ async def test_oauth_with_follow(hass):
"""Test state with oauth and follow."""
twitch_mock = MagicMock()
twitch_mock.users.translate_usernames_to_ids.return_value = [USER_ID]
twitch_mock.channels.get_by_id.return_value = CHANNEL_OBJECT
twitch_mock._oauth_token = True # A replacement for the token
twitch_mock.users.get.return_value = OAUTH_USER_ID
twitch_mock.users.check_subscribed_to_channel.side_effect = HTTPError()
twitch_mock.users.check_follows_channel.return_value = FOLLOW_ACTIVE
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.TwitchClient",
"homeassistant.components.twitch.sensor.Twitch",
return_value=twitch_mock,
):
assert await async_setup_component(hass, sensor.DOMAIN, CONFIG_WITH_OAUTH)