Add config flow to Twitch (#93451)
* Update twitch API * Update twitchAPI * Add tests * Apply suggestions from code review Co-authored-by: Franck Nijhof <frenck@frenck.nl> * Update sensor.py * Update sensor.py * Update sensor.py * Update sensor.py * Update sensor.py * Fix coverage * Move Twitch constants to separate file * Move Twitch constants to separate file * Move Twitch constants to separate file * Add application credentials * Add config flow * Try to add tests * Add strings * Add tests * Add tests * Improve tests * Fix tests * Extract Twitch client creation * Fix reauth * Remove import flow * Remove import flow * Remove reauth * Update * Fix Ruff * Fix feedback * Add strings * Add reauth * Do stuff in init * Fix stuff * Fix stuff * Fix stuff * Fix stuff * Fix stuff * Start with tests * Test coverage * Test coverage * Remove strings * Cleanup * Fix feedback * Fix feedback --------- Co-authored-by: Franck Nijhof <frenck@frenck.nl>
This commit is contained in:
Joost Lekkerkerker
GitHub
parent
91fcbb41b0
commit
25a80cd46f
18 changed files with 1155 additions and 300 deletions
|
|
@ -1 +1,53 @@
|
|||
"""The Twitch component."""
|
||||
from __future__ import annotations
|
||||
|
||||
from aiohttp.client_exceptions import ClientError, ClientResponseError
|
||||
from twitchAPI.twitch import Twitch
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_CLIENT_ID, CONF_TOKEN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
|
||||
from homeassistant.helpers.config_entry_oauth2_flow import (
|
||||
OAuth2Session,
|
||||
async_get_config_entry_implementation,
|
||||
)
|
||||
|
||||
from .const import DOMAIN, OAUTH_SCOPES, PLATFORMS
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up Twitch from a config entry."""
|
||||
implementation = await async_get_config_entry_implementation(hass, entry)
|
||||
session = OAuth2Session(hass, entry, implementation)
|
||||
try:
|
||||
await session.async_ensure_token_valid()
|
||||
except ClientResponseError as err:
|
||||
if 400 <= err.status < 500:
|
||||
raise ConfigEntryAuthFailed(
|
||||
"OAuth session is not valid, reauth required"
|
||||
) from err
|
||||
raise ConfigEntryNotReady from err
|
||||
except ClientError as err:
|
||||
raise ConfigEntryNotReady from err
|
||||
|
||||
app_id = implementation.__dict__[CONF_CLIENT_ID]
|
||||
access_token = entry.data[CONF_TOKEN][CONF_ACCESS_TOKEN]
|
||||
client = await Twitch(
|
||||
app_id=app_id,
|
||||
authenticate_app=False,
|
||||
)
|
||||
client.auto_refresh_auth = False
|
||||
await client.set_user_authentication(access_token, scope=OAUTH_SCOPES)
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = client
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload Twitch config entry."""
|
||||
|
||||
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
|
|
|||
14
homeassistant/components/twitch/application_credentials.py
Normal file
14
homeassistant/components/twitch/application_credentials.py
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
"""application_credentials platform the Twitch integration."""
|
||||
|
||||
from homeassistant.components.application_credentials import AuthorizationServer
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import OAUTH2_AUTHORIZE, OAUTH2_TOKEN
|
||||
|
||||
|
||||
async def async_get_authorization_server(_: HomeAssistant) -> AuthorizationServer:
|
||||
"""Return authorization server."""
|
||||
return AuthorizationServer(
|
||||
authorize_url=OAUTH2_AUTHORIZE,
|
||||
token_url=OAUTH2_TOKEN,
|
||||
)
|
||||
189
homeassistant/components/twitch/config_flow.py
Normal file
189
homeassistant/components/twitch/config_flow.py
Normal file
|
|
@ -0,0 +1,189 @@
|
|||
"""Config flow for Twitch."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from twitchAPI.helper import first
|
||||
from twitchAPI.twitch import Twitch
|
||||
from twitchAPI.type import AuthScope, InvalidTokenException
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_CLIENT_ID, CONF_TOKEN
|
||||
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN
|
||||
from homeassistant.data_entry_flow import AbortFlow, FlowResult
|
||||
from homeassistant.helpers import config_entry_oauth2_flow
|
||||
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
|
||||
|
||||
from .const import CONF_CHANNELS, CONF_REFRESH_TOKEN, DOMAIN, LOGGER, OAUTH_SCOPES
|
||||
|
||||
|
||||
class OAuth2FlowHandler(
|
||||
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
|
||||
):
|
||||
"""Config flow to handle Twitch OAuth2 authentication."""
|
||||
|
||||
DOMAIN = DOMAIN
|
||||
reauth_entry: ConfigEntry | None = None
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize flow."""
|
||||
super().__init__()
|
||||
self.data: dict[str, Any] = {}
|
||||
|
||||
@property
|
||||
def logger(self) -> logging.Logger:
|
||||
"""Return logger."""
|
||||
return LOGGER
|
||||
|
||||
@property
|
||||
def extra_authorize_data(self) -> dict[str, Any]:
|
||||
"""Extra data that needs to be appended to the authorize url."""
|
||||
return {"scope": " ".join([scope.value for scope in OAUTH_SCOPES])}
|
||||
|
||||
async def async_oauth_create_entry(
|
||||
self,
|
||||
data: dict[str, Any],
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
|
||||
client = await Twitch(
|
||||
app_id=self.flow_impl.__dict__[CONF_CLIENT_ID],
|
||||
authenticate_app=False,
|
||||
)
|
||||
client.auto_refresh_auth = False
|
||||
await client.set_user_authentication(
|
||||
data[CONF_TOKEN][CONF_ACCESS_TOKEN], scope=OAUTH_SCOPES
|
||||
)
|
||||
user = await first(client.get_users())
|
||||
assert user
|
||||
|
||||
user_id = user.id
|
||||
|
||||
if not self.reauth_entry:
|
||||
await self.async_set_unique_id(user_id)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
channels = [
|
||||
channel.broadcaster_login
|
||||
async for channel in await client.get_followed_channels(user_id)
|
||||
]
|
||||
|
||||
return self.async_create_entry(
|
||||
title=user.display_name, data=data, options={CONF_CHANNELS: channels}
|
||||
)
|
||||
|
||||
if self.reauth_entry.unique_id == user_id:
|
||||
new_channels = self.reauth_entry.options[CONF_CHANNELS]
|
||||
# Since we could not get all channels at import, we do it at the reauth
|
||||
# immediately after.
|
||||
if "imported" in self.reauth_entry.data:
|
||||
channels = [
|
||||
channel.broadcaster_login
|
||||
async for channel in await client.get_followed_channels(user_id)
|
||||
]
|
||||
options = list(set(channels) - set(new_channels))
|
||||
new_channels = [*new_channels, *options]
|
||||
|
||||
self.hass.config_entries.async_update_entry(
|
||||
self.reauth_entry,
|
||||
data=data,
|
||||
options={CONF_CHANNELS: new_channels},
|
||||
)
|
||||
await self.hass.config_entries.async_reload(self.reauth_entry.entry_id)
|
||||
return self.async_abort(reason="reauth_successful")
|
||||
|
||||
return self.async_abort(
|
||||
reason="wrong_account",
|
||||
description_placeholders={"title": self.reauth_entry.title},
|
||||
)
|
||||
|
||||
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
|
||||
"""Perform reauth upon an API authentication error."""
|
||||
self.reauth_entry = self.hass.config_entries.async_get_entry(
|
||||
self.context["entry_id"]
|
||||
)
|
||||
return await self.async_step_reauth_confirm()
|
||||
|
||||
async def async_step_reauth_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Confirm reauth dialog."""
|
||||
if user_input is None:
|
||||
return self.async_show_form(step_id="reauth_confirm")
|
||||
return await self.async_step_user()
|
||||
|
||||
async def async_step_import(self, config: dict[str, Any]) -> FlowResult:
|
||||
"""Import from yaml."""
|
||||
client = await Twitch(
|
||||
app_id=config[CONF_CLIENT_ID],
|
||||
authenticate_app=False,
|
||||
)
|
||||
client.auto_refresh_auth = False
|
||||
token = config[CONF_TOKEN]
|
||||
try:
|
||||
await client.set_user_authentication(
|
||||
token, validate=True, scope=[AuthScope.USER_READ_SUBSCRIPTIONS]
|
||||
)
|
||||
except InvalidTokenException:
|
||||
async_create_issue(
|
||||
self.hass,
|
||||
DOMAIN,
|
||||
"deprecated_yaml_invalid_token",
|
||||
breaks_in_ha_version="2024.4.0",
|
||||
is_fixable=False,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml_invalid_token",
|
||||
translation_placeholders={
|
||||
"domain": DOMAIN,
|
||||
"integration_title": "Twitch",
|
||||
},
|
||||
)
|
||||
return self.async_abort(reason="invalid_token")
|
||||
user = await first(client.get_users())
|
||||
assert user
|
||||
await self.async_set_unique_id(user.id)
|
||||
try:
|
||||
self._abort_if_unique_id_configured()
|
||||
except AbortFlow as err:
|
||||
async_create_issue(
|
||||
self.hass,
|
||||
DOMAIN,
|
||||
"deprecated_yaml_already_imported",
|
||||
breaks_in_ha_version="2024.4.0",
|
||||
is_fixable=False,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml_already_imported",
|
||||
translation_placeholders={
|
||||
"domain": DOMAIN,
|
||||
"integration_title": "Twitch",
|
||||
},
|
||||
)
|
||||
raise err
|
||||
async_create_issue(
|
||||
self.hass,
|
||||
HOMEASSISTANT_DOMAIN,
|
||||
"deprecated_yaml",
|
||||
breaks_in_ha_version="2024.4.0",
|
||||
is_fixable=False,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml",
|
||||
translation_placeholders={
|
||||
"domain": DOMAIN,
|
||||
"integration_title": "Twitch",
|
||||
},
|
||||
)
|
||||
return self.async_create_entry(
|
||||
title=user.display_name,
|
||||
data={
|
||||
"auth_implementation": DOMAIN,
|
||||
CONF_TOKEN: {
|
||||
CONF_ACCESS_TOKEN: token,
|
||||
CONF_REFRESH_TOKEN: "",
|
||||
"expires_at": 0,
|
||||
},
|
||||
"imported": True,
|
||||
},
|
||||
options={CONF_CHANNELS: config[CONF_CHANNELS]},
|
||||
)
|
||||
|
|
@ -3,8 +3,18 @@ import logging
|
|||
|
||||
from twitchAPI.twitch import AuthScope
|
||||
|
||||
from homeassistant.const import Platform
|
||||
|
||||
LOGGER = logging.getLogger(__package__)
|
||||
|
||||
PLATFORMS = [Platform.SENSOR]
|
||||
|
||||
OAUTH2_AUTHORIZE = "https://id.twitch.tv/oauth2/authorize"
|
||||
OAUTH2_TOKEN = "https://id.twitch.tv/oauth2/token"
|
||||
|
||||
CONF_REFRESH_TOKEN = "refresh_token"
|
||||
|
||||
DOMAIN = "twitch"
|
||||
CONF_CHANNELS = "channels"
|
||||
|
||||
OAUTH_SCOPES = [AuthScope.USER_READ_SUBSCRIPTIONS]
|
||||
OAUTH_SCOPES = [AuthScope.USER_READ_SUBSCRIPTIONS, AuthScope.USER_READ_FOLLOWS]
|
||||
|
|
|
|||
|
|
@ -2,8 +2,10 @@
|
|||
"domain": "twitch",
|
||||
"name": "Twitch",
|
||||
"codeowners": ["@joostlek"],
|
||||
"config_flow": true,
|
||||
"dependencies": ["application_credentials"],
|
||||
"documentation": "https://www.home-assistant.io/integrations/twitch",
|
||||
"iot_class": "cloud_polling",
|
||||
"loggers": ["twitch"],
|
||||
"requirements": ["twitchAPI==3.10.0"]
|
||||
"requirements": ["twitchAPI==4.0.0"]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,24 +4,27 @@ from __future__ import annotations
|
|||
from twitchAPI.helper import first
|
||||
from twitchAPI.twitch import (
|
||||
AuthType,
|
||||
InvalidTokenException,
|
||||
MissingScopeException,
|
||||
Twitch,
|
||||
TwitchAPIException,
|
||||
TwitchAuthorizationException,
|
||||
TwitchResourceNotFound,
|
||||
TwitchUser,
|
||||
)
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.application_credentials import (
|
||||
ClientCredential,
|
||||
async_import_client_credential,
|
||||
)
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
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
|
||||
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from .const import CONF_CHANNELS, LOGGER, OAUTH_SCOPES
|
||||
from .const import CONF_CHANNELS, DOMAIN, LOGGER, OAUTH_SCOPES
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
|
|
@ -56,40 +59,46 @@ async def async_setup_platform(
|
|||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""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)
|
||||
|
||||
try:
|
||||
client = await Twitch(
|
||||
app_id=client_id,
|
||||
app_secret=client_secret,
|
||||
target_app_auth_scope=OAUTH_SCOPES,
|
||||
await async_import_client_credential(
|
||||
hass,
|
||||
DOMAIN,
|
||||
ClientCredential(config[CONF_CLIENT_ID], config[CONF_CLIENT_SECRET]),
|
||||
)
|
||||
client.auto_refresh_auth = False
|
||||
except TwitchAuthorizationException:
|
||||
LOGGER.error("Invalid client ID or client secret")
|
||||
return
|
||||
|
||||
if oauth_token:
|
||||
try:
|
||||
await client.set_user_authentication(
|
||||
token=oauth_token, scope=OAUTH_SCOPES, validate=True
|
||||
if CONF_TOKEN in config:
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
|
||||
)
|
||||
)
|
||||
else:
|
||||
async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
"deprecated_yaml_credentials_imported",
|
||||
breaks_in_ha_version="2024.4.0",
|
||||
is_fixable=False,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml_credentials_imported",
|
||||
translation_placeholders={
|
||||
"domain": DOMAIN,
|
||||
"integration_title": "Twitch",
|
||||
},
|
||||
)
|
||||
except MissingScopeException:
|
||||
LOGGER.error("OAuth token is missing required scope")
|
||||
return
|
||||
except InvalidTokenException:
|
||||
LOGGER.error("OAuth token is invalid")
|
||||
return
|
||||
|
||||
twitch_users: list[TwitchUser] = []
|
||||
async for channel in client.get_users(logins=channels):
|
||||
twitch_users.append(channel)
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Initialize entries."""
|
||||
client = hass.data[DOMAIN][entry.entry_id]
|
||||
|
||||
async_add_entities(
|
||||
[TwitchSensor(channel, client) for channel in twitch_users],
|
||||
[
|
||||
TwitchSensor(channel, client)
|
||||
async for channel in client.get_users(logins=entry.options[CONF_CHANNELS])
|
||||
],
|
||||
True,
|
||||
)
|
||||
|
||||
|
|
@ -109,7 +118,7 @@ class TwitchSensor(SensorEntity):
|
|||
|
||||
async def async_update(self) -> None:
|
||||
"""Update device state."""
|
||||
followers = (await self._client.get_users_follows(to_id=self._channel.id)).total
|
||||
followers = (await self._client.get_channel_followers(self._channel.id)).total
|
||||
self._attr_extra_state_attributes = {
|
||||
ATTR_FOLLOWING: followers,
|
||||
ATTR_VIEWS: self._channel.view_count,
|
||||
|
|
@ -149,13 +158,11 @@ class TwitchSensor(SensorEntity):
|
|||
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
|
||||
follows = await self._client.get_followed_channels(
|
||||
user.id, broadcaster_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[
|
||||
self._attr_extra_state_attributes[ATTR_FOLLOW] = follows.total > 0
|
||||
if follows.total:
|
||||
self._attr_extra_state_attributes[ATTR_FOLLOW_SINCE] = follows.data[
|
||||
0
|
||||
].followed_at
|
||||
|
|
|
|||
30
homeassistant/components/twitch/strings.json
Normal file
30
homeassistant/components/twitch/strings.json
Normal file
|
|
@ -0,0 +1,30 @@
|
|||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"reauth_confirm": {
|
||||
"title": "[%key:common::config_flow::title::reauth%]",
|
||||
"description": "The Twitch integration needs to re-authenticate your account"
|
||||
}
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
|
||||
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]",
|
||||
"wrong_account": "Wrong account: Please authenticate with {username}."
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
"deprecated_yaml_invalid_token": {
|
||||
"title": "The {integration_title} YAML configuration is being removed",
|
||||
"description": "Configuring {integration_title} using YAML is being removed.\n\nYour configuration couldn't be imported because the token in the configuration.yaml was invalid.\n\nPlease add Twitch again via the UI.\n\nRemove the `{domain}` configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
|
||||
},
|
||||
"deprecated_yaml_credentials_imported": {
|
||||
"title": "The {integration_title} YAML configuration is being removed",
|
||||
"description": "Configuring {integration_title} using YAML is being removed.\n\nYour application credentials are imported, but a config entry could not be created because there was no access token.\n\nPlease add Twitch again via the UI.\n\nRemove the `{domain}` configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
|
||||
},
|
||||
"deprecated_yaml_already_imported": {
|
||||
"title": "The {integration_title} YAML configuration is being removed",
|
||||
"description": "Configuring {integration_title} using YAML is being removed.\n\nYour application credentials are already imported.\n\nRemove the `{domain}` configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -18,6 +18,7 @@ APPLICATION_CREDENTIALS = [
|
|||
"netatmo",
|
||||
"senz",
|
||||
"spotify",
|
||||
"twitch",
|
||||
"withings",
|
||||
"xbox",
|
||||
"yolink",
|
||||
|
|
|
|||
|
|
@ -500,6 +500,7 @@ FLOWS = {
|
|||
"twentemilieu",
|
||||
"twilio",
|
||||
"twinkly",
|
||||
"twitch",
|
||||
"ukraine_alarm",
|
||||
"unifi",
|
||||
"unifiprotect",
|
||||
|
|
|
|||
|
|
@ -6021,7 +6021,7 @@
|
|||
"twitch": {
|
||||
"name": "Twitch",
|
||||
"integration_type": "hub",
|
||||
"config_flow": false,
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_polling"
|
||||
},
|
||||
"twitter": {
|
||||
|
|
|
|||
|
|
@ -2613,7 +2613,7 @@ twentemilieu==1.0.0
|
|||
twilio==6.32.0
|
||||
|
||||
# homeassistant.components.twitch
|
||||
twitchAPI==3.10.0
|
||||
twitchAPI==4.0.0
|
||||
|
||||
# homeassistant.components.ukraine_alarm
|
||||
uasiren==0.0.1
|
||||
|
|
|
|||
|
|
@ -1934,7 +1934,7 @@ twentemilieu==1.0.0
|
|||
twilio==6.32.0
|
||||
|
||||
# homeassistant.components.twitch
|
||||
twitchAPI==3.10.0
|
||||
twitchAPI==4.0.0
|
||||
|
||||
# homeassistant.components.ukraine_alarm
|
||||
uasiren==0.0.1
|
||||
|
|
|
|||
|
|
@ -1,10 +1,10 @@
|
|||
"""Tests for the Twitch component."""
|
||||
import asyncio
|
||||
from collections.abc import AsyncGenerator
|
||||
from collections.abc import AsyncGenerator, AsyncIterator
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
from datetime import datetime
|
||||
|
||||
from twitchAPI.object import TwitchUser
|
||||
from twitchAPI.object.api import FollowedChannelsResult, TwitchUser
|
||||
from twitchAPI.twitch import (
|
||||
InvalidTokenException,
|
||||
MissingScopeException,
|
||||
|
|
@ -12,24 +12,34 @@ from twitchAPI.twitch import (
|
|||
TwitchAuthorizationException,
|
||||
TwitchResourceNotFound,
|
||||
)
|
||||
from twitchAPI.types import AuthScope, AuthType
|
||||
from twitchAPI.type import AuthScope, AuthType
|
||||
|
||||
USER_OBJECT: TwitchUser = TwitchUser(
|
||||
id=123,
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
|
||||
"""Fixture for setting up the component."""
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
|
||||
|
||||
def _get_twitch_user(user_id: str = "123") -> TwitchUser:
|
||||
return TwitchUser(
|
||||
id=user_id,
|
||||
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
|
||||
async def async_iterator(iterable) -> AsyncIterator:
|
||||
"""Return async iterator."""
|
||||
for i in iterable:
|
||||
yield i
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
@ -41,12 +51,20 @@ class UserSubscriptionMock:
|
|||
|
||||
|
||||
@dataclass
|
||||
class UserFollowMock:
|
||||
"""User follow mock."""
|
||||
class FollowedChannelMock:
|
||||
"""Followed channel mock."""
|
||||
|
||||
broadcaster_login: str
|
||||
followed_at: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelFollowerMock:
|
||||
"""Channel follower mock."""
|
||||
|
||||
user_id: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class StreamMock:
|
||||
"""Stream mock."""
|
||||
|
|
@ -56,6 +74,32 @@ class StreamMock:
|
|||
thumbnail_url: str
|
||||
|
||||
|
||||
class TwitchUserFollowResultMock:
|
||||
"""Mock for twitch user follow result."""
|
||||
|
||||
def __init__(self, follows: list[FollowedChannelMock]) -> None:
|
||||
"""Initialize mock."""
|
||||
self.total = len(follows)
|
||||
self.data = follows
|
||||
|
||||
def __aiter__(self):
|
||||
"""Return async iterator."""
|
||||
return async_iterator(self.data)
|
||||
|
||||
|
||||
class ChannelFollowersResultMock:
|
||||
"""Mock for twitch channel follow result."""
|
||||
|
||||
def __init__(self, follows: list[ChannelFollowerMock]) -> None:
|
||||
"""Initialize mock."""
|
||||
self.total = len(follows)
|
||||
self.data = follows
|
||||
|
||||
def __aiter__(self):
|
||||
"""Return async iterator."""
|
||||
return async_iterator(self.data)
|
||||
|
||||
|
||||
STREAMS = StreamMock(
|
||||
game_name="Good game", title="Title", thumbnail_url="stream-medium.png"
|
||||
)
|
||||
|
|
@ -64,25 +108,18 @@ STREAMS = StreamMock(
|
|||
class TwitchMock:
|
||||
"""Mock for the twitch object."""
|
||||
|
||||
is_streaming = True
|
||||
is_gifted = False
|
||||
is_subscribed = False
|
||||
is_following = True
|
||||
different_user_id = False
|
||||
|
||||
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
|
||||
|
|
@ -91,7 +128,8 @@ class TwitchMock:
|
|||
self, user_ids: list[str] | None = None, logins: list[str] | None = None
|
||||
) -> AsyncGenerator[TwitchUser, None]:
|
||||
"""Get list of mock users."""
|
||||
for user in [USER_OBJECT]:
|
||||
users = [_get_twitch_user("234" if self.different_user_id else "123")]
|
||||
for user in users:
|
||||
yield user
|
||||
|
||||
def has_required_auth(
|
||||
|
|
@ -100,38 +138,56 @@ class TwitchMock:
|
|||
"""Return if auth required."""
|
||||
return True
|
||||
|
||||
async def get_users_follows(
|
||||
self, to_id: str | None = None, from_id: str | None = 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:
|
||||
if self.is_subscribed:
|
||||
return UserSubscriptionMock(
|
||||
broadcaster_id=broadcaster_id, is_gift=self._is_gifted
|
||||
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
|
||||
self,
|
||||
token: str,
|
||||
scope: list[AuthScope],
|
||||
validate: bool = True,
|
||||
) -> None:
|
||||
"""Set user authentication."""
|
||||
pass
|
||||
|
||||
async def get_followed_channels(
|
||||
self, user_id: str, broadcaster_id: str | None = None
|
||||
) -> FollowedChannelsResult:
|
||||
"""Get followed channels."""
|
||||
if self.is_following:
|
||||
return TwitchUserFollowResultMock(
|
||||
[
|
||||
FollowedChannelMock(
|
||||
followed_at=datetime(year=2023, month=8, day=1),
|
||||
broadcaster_login="internetofthings",
|
||||
),
|
||||
FollowedChannelMock(
|
||||
followed_at=datetime(year=2023, month=8, day=1),
|
||||
broadcaster_login="homeassistant",
|
||||
),
|
||||
]
|
||||
)
|
||||
return TwitchUserFollowResultMock([])
|
||||
|
||||
async def get_channel_followers(
|
||||
self, broadcaster_id: str
|
||||
) -> ChannelFollowersResultMock:
|
||||
"""Get channel followers."""
|
||||
return ChannelFollowersResultMock([ChannelFollowerMock(user_id="abc")])
|
||||
|
||||
async def get_streams(
|
||||
self, user_id: list[str], first: int
|
||||
) -> AsyncGenerator[StreamMock, None]:
|
||||
"""Get streams for the user."""
|
||||
streams = []
|
||||
if self._is_streaming:
|
||||
if self.is_streaming:
|
||||
streams = [STREAMS]
|
||||
for stream in streams:
|
||||
yield stream
|
||||
|
|
|
|||
110
tests/components/twitch/conftest.py
Normal file
110
tests/components/twitch/conftest.py
Normal file
|
|
@ -0,0 +1,110 @@
|
|||
"""Configure tests for the Twitch integration."""
|
||||
from collections.abc import Awaitable, Callable, Generator
|
||||
import time
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.application_credentials import (
|
||||
ClientCredential,
|
||||
async_import_client_credential,
|
||||
)
|
||||
from homeassistant.components.twitch.const import DOMAIN, OAUTH2_TOKEN, OAUTH_SCOPES
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.components.twitch import TwitchMock
|
||||
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||
|
||||
ComponentSetup = Callable[[TwitchMock | None], Awaitable[None]]
|
||||
|
||||
CLIENT_ID = "1234"
|
||||
CLIENT_SECRET = "5678"
|
||||
TITLE = "Test"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
|
||||
"""Override async_setup_entry."""
|
||||
with patch(
|
||||
"homeassistant.components.twitch.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture(name="scopes")
|
||||
def mock_scopes() -> list[str]:
|
||||
"""Fixture to set the scopes present in the OAuth token."""
|
||||
return [scope.value for scope in OAUTH_SCOPES]
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def setup_credentials(hass: HomeAssistant) -> None:
|
||||
"""Fixture to setup credentials."""
|
||||
assert await async_setup_component(hass, "application_credentials", {})
|
||||
await async_import_client_credential(
|
||||
hass,
|
||||
DOMAIN,
|
||||
ClientCredential(CLIENT_ID, CLIENT_SECRET),
|
||||
DOMAIN,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(name="expires_at")
|
||||
def mock_expires_at() -> int:
|
||||
"""Fixture to set the oauth token expiration time."""
|
||||
return time.time() + 3600
|
||||
|
||||
|
||||
@pytest.fixture(name="config_entry")
|
||||
def mock_config_entry(expires_at: int, scopes: list[str]) -> MockConfigEntry:
|
||||
"""Create Twitch entry in Home Assistant."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title=TITLE,
|
||||
unique_id="123",
|
||||
data={
|
||||
"auth_implementation": DOMAIN,
|
||||
"token": {
|
||||
"access_token": "mock-access-token",
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"expires_at": expires_at,
|
||||
"scope": " ".join(scopes),
|
||||
},
|
||||
},
|
||||
options={"channels": ["internetofthings"]},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_connection(aioclient_mock: AiohttpClientMocker) -> None:
|
||||
"""Mock Twitch connection."""
|
||||
aioclient_mock.post(
|
||||
OAUTH2_TOKEN,
|
||||
json={
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"access_token": "mock-access-token",
|
||||
"type": "Bearer",
|
||||
"expires_in": 60,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(name="twitch_mock")
|
||||
def twitch_mock() -> TwitchMock:
|
||||
"""Return as fixture to inject other mocks."""
|
||||
return TwitchMock()
|
||||
|
||||
|
||||
@pytest.fixture(name="twitch")
|
||||
def mock_twitch(twitch_mock: TwitchMock):
|
||||
"""Mock Twitch."""
|
||||
with patch(
|
||||
"homeassistant.components.twitch.Twitch",
|
||||
return_value=twitch_mock,
|
||||
), patch(
|
||||
"homeassistant.components.twitch.config_flow.Twitch",
|
||||
return_value=twitch_mock,
|
||||
):
|
||||
yield twitch_mock
|
||||
295
tests/components/twitch/test_config_flow.py
Normal file
295
tests/components/twitch/test_config_flow.py
Normal file
|
|
@ -0,0 +1,295 @@
|
|||
"""Test config flow for Twitch."""
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.twitch.const import (
|
||||
CONF_CHANNELS,
|
||||
DOMAIN,
|
||||
OAUTH2_AUTHORIZE,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_REAUTH, SOURCE_USER
|
||||
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET, CONF_TOKEN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResult, FlowResultType
|
||||
from homeassistant.helpers import config_entry_oauth2_flow, issue_registry as ir
|
||||
|
||||
from . import setup_integration
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.components.twitch import TwitchInvalidTokenMock, TwitchMock
|
||||
from tests.components.twitch.conftest import CLIENT_ID, TITLE
|
||||
from tests.typing import ClientSessionGenerator
|
||||
|
||||
|
||||
async def _do_get_token(
|
||||
hass: HomeAssistant,
|
||||
result: FlowResult,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
scopes: list[str],
|
||||
) -> None:
|
||||
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"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
|
||||
"&redirect_uri=https://example.com/auth/external/callback"
|
||||
f"&state={state}&scope={'+'.join(scopes)}"
|
||||
)
|
||||
|
||||
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"
|
||||
|
||||
|
||||
async def test_full_flow(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
mock_setup_entry,
|
||||
twitch: TwitchMock,
|
||||
scopes: list[str],
|
||||
) -> None:
|
||||
"""Check full flow."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
"twitch", context={"source": SOURCE_USER}
|
||||
)
|
||||
await _do_get_token(hass, result, hass_client_no_auth, scopes)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
|
||||
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
|
||||
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "channel123"
|
||||
assert "result" in result
|
||||
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["result"].unique_id == "123"
|
||||
assert result["options"] == {CONF_CHANNELS: ["internetofthings", "homeassistant"]}
|
||||
|
||||
|
||||
async def test_already_configured(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
config_entry: MockConfigEntry,
|
||||
mock_setup_entry,
|
||||
twitch: TwitchMock,
|
||||
scopes: list[str],
|
||||
) -> None:
|
||||
"""Check flow aborts when account already configured."""
|
||||
await setup_integration(hass, config_entry)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
"twitch", context={"source": SOURCE_USER}
|
||||
)
|
||||
await _do_get_token(hass, result, hass_client_no_auth, scopes)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.twitch.config_flow.Twitch", return_value=TwitchMock()
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_reauth(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
config_entry: MockConfigEntry,
|
||||
mock_setup_entry,
|
||||
twitch: TwitchMock,
|
||||
scopes: list[str],
|
||||
) -> None:
|
||||
"""Check reauth flow."""
|
||||
await setup_integration(hass, config_entry)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={
|
||||
"source": SOURCE_REAUTH,
|
||||
"entry_id": config_entry.entry_id,
|
||||
},
|
||||
data=config_entry.data,
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "reauth_confirm"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
||||
|
||||
await _do_get_token(hass, result, hass_client_no_auth, scopes)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "reauth_successful"
|
||||
|
||||
|
||||
async def test_reauth_from_import(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
mock_setup_entry,
|
||||
twitch: TwitchMock,
|
||||
expires_at,
|
||||
scopes: list[str],
|
||||
) -> None:
|
||||
"""Check reauth flow."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title=TITLE,
|
||||
unique_id="123",
|
||||
data={
|
||||
"auth_implementation": DOMAIN,
|
||||
"token": {
|
||||
"access_token": "mock-access-token",
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"expires_at": expires_at,
|
||||
"scope": " ".join(scopes),
|
||||
},
|
||||
"imported": True,
|
||||
},
|
||||
options={"channels": ["internetofthings"]},
|
||||
)
|
||||
await test_reauth(
|
||||
hass,
|
||||
hass_client_no_auth,
|
||||
current_request_with_host,
|
||||
config_entry,
|
||||
mock_setup_entry,
|
||||
twitch,
|
||||
scopes,
|
||||
)
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
entry = entries[0]
|
||||
assert "imported" not in entry.data
|
||||
assert entry.options == {CONF_CHANNELS: ["internetofthings", "homeassistant"]}
|
||||
|
||||
|
||||
async def test_reauth_wrong_account(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
config_entry: MockConfigEntry,
|
||||
mock_setup_entry,
|
||||
twitch: TwitchMock,
|
||||
scopes: list[str],
|
||||
) -> None:
|
||||
"""Check reauth flow."""
|
||||
await setup_integration(hass, config_entry)
|
||||
twitch.different_user_id = True
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={
|
||||
"source": SOURCE_REAUTH,
|
||||
"entry_id": config_entry.entry_id,
|
||||
},
|
||||
data=config_entry.data,
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "reauth_confirm"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
||||
|
||||
await _do_get_token(hass, result, hass_client_no_auth, scopes)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "wrong_account"
|
||||
|
||||
|
||||
async def test_import(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
mock_setup_entry,
|
||||
twitch: TwitchMock,
|
||||
) -> None:
|
||||
"""Test import flow."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={
|
||||
"source": SOURCE_IMPORT,
|
||||
},
|
||||
data={
|
||||
"platform": "twitch",
|
||||
CONF_CLIENT_ID: "1234",
|
||||
CONF_CLIENT_SECRET: "abcd",
|
||||
CONF_TOKEN: "efgh",
|
||||
"channels": ["channel123"],
|
||||
},
|
||||
)
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "channel123"
|
||||
assert "result" in result
|
||||
assert "token" in result["result"].data
|
||||
assert result["result"].data["token"]["access_token"] == "efgh"
|
||||
assert result["result"].data["token"]["refresh_token"] == ""
|
||||
assert result["result"].unique_id == "123"
|
||||
assert result["options"] == {CONF_CHANNELS: ["channel123"]}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("twitch_mock", [TwitchInvalidTokenMock()])
|
||||
async def test_import_invalid_token(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
mock_setup_entry,
|
||||
twitch: TwitchMock,
|
||||
) -> None:
|
||||
"""Test import flow."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={
|
||||
"source": SOURCE_IMPORT,
|
||||
},
|
||||
data={
|
||||
"platform": "twitch",
|
||||
CONF_CLIENT_ID: "1234",
|
||||
CONF_CLIENT_SECRET: "abcd",
|
||||
CONF_TOKEN: "efgh",
|
||||
"channels": ["channel123"],
|
||||
},
|
||||
)
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "invalid_token"
|
||||
issue_registry = ir.async_get(hass)
|
||||
assert len(issue_registry.issues) == 1
|
||||
|
||||
|
||||
async def test_import_already_imported(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
config_entry: MockConfigEntry,
|
||||
mock_setup_entry,
|
||||
twitch: TwitchMock,
|
||||
) -> None:
|
||||
"""Test import flow where the config is already imported."""
|
||||
await setup_integration(hass, config_entry)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={
|
||||
"source": SOURCE_IMPORT,
|
||||
},
|
||||
data={
|
||||
"platform": "twitch",
|
||||
CONF_CLIENT_ID: "1234",
|
||||
CONF_CLIENT_SECRET: "abcd",
|
||||
CONF_TOKEN: "efgh",
|
||||
"channels": ["channel123"],
|
||||
},
|
||||
)
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
issue_registry = ir.async_get(hass)
|
||||
assert len(issue_registry.issues) == 1
|
||||
116
tests/components/twitch/test_init.py
Normal file
116
tests/components/twitch/test_init.py
Normal file
|
|
@ -0,0 +1,116 @@
|
|||
"""Tests for YouTube."""
|
||||
import http
|
||||
import time
|
||||
from unittest.mock import patch
|
||||
|
||||
from aiohttp.client_exceptions import ClientError
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.twitch.const import DOMAIN, OAUTH2_TOKEN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import TwitchMock, setup_integration
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||
|
||||
|
||||
async def test_setup_success(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry, twitch: TwitchMock
|
||||
) -> None:
|
||||
"""Test successful setup and unload."""
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 1
|
||||
assert entries[0].state is ConfigEntryState.LOADED
|
||||
|
||||
await hass.config_entries.async_unload(entries[0].entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert not hass.services.async_services().get(DOMAIN)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("expires_at", [time.time() - 3600], ids=["expired"])
|
||||
async def test_expired_token_refresh_success(
|
||||
hass: HomeAssistant,
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
config_entry: MockConfigEntry,
|
||||
twitch: TwitchMock,
|
||||
) -> None:
|
||||
"""Test expired token is refreshed."""
|
||||
|
||||
aioclient_mock.clear_requests()
|
||||
aioclient_mock.post(
|
||||
OAUTH2_TOKEN,
|
||||
json={
|
||||
"access_token": "updated-access-token",
|
||||
"refresh_token": "updated-refresh-token",
|
||||
"expires_at": time.time() + 3600,
|
||||
"expires_in": 3600,
|
||||
},
|
||||
)
|
||||
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 1
|
||||
assert entries[0].state is ConfigEntryState.LOADED
|
||||
assert entries[0].data["token"]["access_token"] == "updated-access-token"
|
||||
assert entries[0].data["token"]["expires_in"] == 3600
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("expires_at", "status", "expected_state"),
|
||||
[
|
||||
(
|
||||
time.time() - 3600,
|
||||
http.HTTPStatus.UNAUTHORIZED,
|
||||
ConfigEntryState.SETUP_ERROR,
|
||||
),
|
||||
(
|
||||
time.time() - 3600,
|
||||
http.HTTPStatus.INTERNAL_SERVER_ERROR,
|
||||
ConfigEntryState.SETUP_RETRY,
|
||||
),
|
||||
],
|
||||
ids=["failure_requires_reauth", "transient_failure"],
|
||||
)
|
||||
async def test_expired_token_refresh_failure(
|
||||
hass: HomeAssistant,
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
status: http.HTTPStatus,
|
||||
expected_state: ConfigEntryState,
|
||||
config_entry: MockConfigEntry,
|
||||
twitch: TwitchMock,
|
||||
) -> None:
|
||||
"""Test failure while refreshing token with a transient error."""
|
||||
|
||||
aioclient_mock.clear_requests()
|
||||
aioclient_mock.post(
|
||||
OAUTH2_TOKEN,
|
||||
status=status,
|
||||
)
|
||||
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
# Verify a transient failure has occurred
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert entries[0].state is expected_state
|
||||
|
||||
|
||||
async def test_expired_token_refresh_client_error(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry, twitch: TwitchMock
|
||||
) -> None:
|
||||
"""Test failure while refreshing token with a client error."""
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.twitch.OAuth2Session.async_ensure_token_valid",
|
||||
side_effect=ClientError,
|
||||
):
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
# Verify a transient failure has occurred
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert entries[0].state is ConfigEntryState.SETUP_RETRY
|
||||
177
tests/components/twitch/test_sensor.py
Normal file
177
tests/components/twitch/test_sensor.py
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
"""The tests for an update of the Twitch component."""
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
from homeassistant.components.twitch.const import CONF_CHANNELS, DOMAIN
|
||||
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET, CONF_TOKEN, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import issue_registry as ir
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from ...common import MockConfigEntry
|
||||
from . import (
|
||||
TwitchAPIExceptionMock,
|
||||
TwitchInvalidTokenMock,
|
||||
TwitchInvalidUserMock,
|
||||
TwitchMissingScopeMock,
|
||||
TwitchMock,
|
||||
TwitchUnauthorizedMock,
|
||||
setup_integration,
|
||||
)
|
||||
|
||||
ENTITY_ID = "sensor.channel123"
|
||||
CONFIG = {
|
||||
"auth_implementation": "cred",
|
||||
CONF_CLIENT_ID: "1234",
|
||||
CONF_CLIENT_SECRET: "abcd",
|
||||
}
|
||||
|
||||
LEGACY_CONFIG_WITHOUT_TOKEN = {
|
||||
SENSOR_DOMAIN: {
|
||||
"platform": "twitch",
|
||||
CONF_CLIENT_ID: "1234",
|
||||
CONF_CLIENT_SECRET: "abcd",
|
||||
"channels": ["channel123"],
|
||||
}
|
||||
}
|
||||
|
||||
LEGACY_CONFIG = {
|
||||
SENSOR_DOMAIN: {
|
||||
"platform": "twitch",
|
||||
CONF_CLIENT_ID: "1234",
|
||||
CONF_CLIENT_SECRET: "abcd",
|
||||
CONF_TOKEN: "efgh",
|
||||
"channels": ["channel123"],
|
||||
}
|
||||
}
|
||||
|
||||
OPTIONS = {CONF_CHANNELS: ["channel123"]}
|
||||
|
||||
|
||||
async def test_legacy_migration(
|
||||
hass: HomeAssistant, twitch: TwitchMock, mock_setup_entry
|
||||
) -> None:
|
||||
"""Test importing legacy yaml."""
|
||||
assert await async_setup_component(hass, Platform.SENSOR, LEGACY_CONFIG)
|
||||
await hass.async_block_till_done()
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 1
|
||||
issue_registry = ir.async_get(hass)
|
||||
assert len(issue_registry.issues) == 1
|
||||
|
||||
|
||||
async def test_legacy_migration_without_token(
|
||||
hass: HomeAssistant, twitch: TwitchMock
|
||||
) -> None:
|
||||
"""Test importing legacy yaml."""
|
||||
assert await async_setup_component(
|
||||
hass, Platform.SENSOR, LEGACY_CONFIG_WITHOUT_TOKEN
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 0
|
||||
issue_registry = ir.async_get(hass)
|
||||
assert len(issue_registry.issues) == 1
|
||||
|
||||
|
||||
async def test_offline(
|
||||
hass: HomeAssistant, twitch: TwitchMock, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test offline state."""
|
||||
twitch.is_streaming = False
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
sensor_state = hass.states.get(ENTITY_ID)
|
||||
assert sensor_state.state == "offline"
|
||||
assert sensor_state.attributes["entity_picture"] == "logo.png"
|
||||
|
||||
|
||||
async def test_streaming(
|
||||
hass: HomeAssistant, twitch: TwitchMock, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test streaming state."""
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
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["title"] == "Title"
|
||||
|
||||
|
||||
async def test_oauth_without_sub_and_follow(
|
||||
hass: HomeAssistant, twitch: TwitchMock, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test state with oauth."""
|
||||
twitch.is_following = False
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
sensor_state = hass.states.get(ENTITY_ID)
|
||||
assert sensor_state.attributes["subscribed"] is False
|
||||
assert sensor_state.attributes["following"] is False
|
||||
|
||||
|
||||
async def test_oauth_with_sub(
|
||||
hass: HomeAssistant, twitch: TwitchMock, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test state with oauth and sub."""
|
||||
twitch.is_subscribed = True
|
||||
twitch.is_following = False
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
sensor_state = hass.states.get(ENTITY_ID)
|
||||
assert sensor_state.attributes["subscribed"] is True
|
||||
assert sensor_state.attributes["subscription_is_gifted"] is False
|
||||
assert sensor_state.attributes["following"] is False
|
||||
|
||||
|
||||
async def test_oauth_with_follow(
|
||||
hass: HomeAssistant, twitch: TwitchMock, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test state with oauth and follow."""
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
sensor_state = hass.states.get(ENTITY_ID)
|
||||
assert sensor_state.attributes["following"] is True
|
||||
assert sensor_state.attributes["following_since"] == datetime(
|
||||
year=2023, month=8, day=1
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"twitch_mock",
|
||||
[TwitchUnauthorizedMock(), TwitchMissingScopeMock(), TwitchInvalidTokenMock()],
|
||||
)
|
||||
async def test_auth_invalid(
|
||||
hass: HomeAssistant, twitch: TwitchMock, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test auth failures."""
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
sensor_state = hass.states.get(ENTITY_ID)
|
||||
assert sensor_state is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("twitch_mock", [TwitchInvalidUserMock()])
|
||||
async def test_auth_with_invalid_user(
|
||||
hass: HomeAssistant, twitch: TwitchMock, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test auth with invalid user."""
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
sensor_state = hass.states.get(ENTITY_ID)
|
||||
assert "subscribed" not in sensor_state.attributes
|
||||
|
||||
|
||||
@pytest.mark.parametrize("twitch_mock", [TwitchAPIExceptionMock()])
|
||||
async def test_auth_with_api_exception(
|
||||
hass: HomeAssistant, twitch: TwitchMock, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test auth with invalid user."""
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
sensor_state = hass.states.get(ENTITY_ID)
|
||||
assert sensor_state.attributes["subscribed"] is False
|
||||
assert "subscription_is_gifted" not in sensor_state.attributes
|
||||
|
|
@ -1,205 +0,0 @@
|
|||
"""The tests for an update of the Twitch component."""
|
||||
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: {
|
||||
"platform": "twitch",
|
||||
CONF_CLIENT_ID: "1234",
|
||||
CONF_CLIENT_SECRET: " abcd",
|
||||
"channels": ["channel123"],
|
||||
}
|
||||
}
|
||||
CONFIG_WITH_OAUTH = {
|
||||
sensor.DOMAIN: {
|
||||
"platform": "twitch",
|
||||
CONF_CLIENT_ID: "1234",
|
||||
CONF_CLIENT_SECRET: "abcd",
|
||||
"channels": ["channel123"],
|
||||
"token": "9876",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
async def test_init(hass: HomeAssistant) -> None:
|
||||
"""Test initial config."""
|
||||
|
||||
with patch(
|
||||
"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()
|
||||
|
||||
sensor_state = hass.states.get(ENTITY_ID)
|
||||
assert sensor_state.state == "offline"
|
||||
assert sensor_state.name == "channel123"
|
||||
assert sensor_state.attributes["icon"] == "mdi:twitch"
|
||||
assert sensor_state.attributes["friendly_name"] == "channel123"
|
||||
assert sensor_state.attributes["views"] == 42
|
||||
assert sensor_state.attributes["followers"] == 24
|
||||
|
||||
|
||||
async def test_offline(hass: HomeAssistant) -> None:
|
||||
"""Test offline state."""
|
||||
|
||||
with patch(
|
||||
"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()
|
||||
|
||||
sensor_state = hass.states.get(ENTITY_ID)
|
||||
assert sensor_state.state == "offline"
|
||||
assert sensor_state.attributes["entity_picture"] == "logo.png"
|
||||
|
||||
|
||||
async def test_streaming(hass: HomeAssistant) -> None:
|
||||
"""Test streaming state."""
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.twitch.sensor.Twitch",
|
||||
return_value=TwitchMock(),
|
||||
):
|
||||
assert await async_setup_component(hass, sensor.DOMAIN, CONFIG) is True
|
||||
await hass.async_block_till_done()
|
||||
|
||||
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["title"] == "Title"
|
||||
|
||||
|
||||
async def test_oauth_without_sub_and_follow(hass: HomeAssistant) -> None:
|
||||
"""Test state with oauth."""
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.twitch.sensor.Twitch",
|
||||
return_value=TwitchMock(is_following=False),
|
||||
):
|
||||
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 False
|
||||
|
||||
|
||||
async def test_oauth_with_sub(hass: HomeAssistant) -> None:
|
||||
"""Test state with oauth and sub."""
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.twitch.sensor.Twitch",
|
||||
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()
|
||||
|
||||
sensor_state = hass.states.get(ENTITY_ID)
|
||||
assert sensor_state.attributes["subscribed"] is True
|
||||
assert sensor_state.attributes["subscription_is_gifted"] is False
|
||||
assert sensor_state.attributes["following"] is False
|
||||
|
||||
|
||||
async def test_oauth_with_follow(hass: HomeAssistant) -> None:
|
||||
"""Test state with oauth and follow."""
|
||||
|
||||
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=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 "subscription_is_gifted" not in sensor_state.attributes
|
||||
Loading…
Add table
Add a link
Reference in a new issue