Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add ability to subscribe to own YouTube channels (#141693)

pull/141718/head
Franck Nijhof 2025-03-29 00:59:24 +01:00 committed by GitHub
parent 42d6bd3839
commit fcd4d3e2df
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 244 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
homeassistant/components/youtube/config_flow.py
View File

@ -7,7 +7,6 @@ import logging
from typing import Any
import voluptuous as vol
from youtubeaio.helper import first
from youtubeaio.types import AuthScope, ForbiddenError
from youtubeaio.youtube import YouTube
@ -96,8 +95,12 @@ class OAuth2FlowHandler(
"""Create an entry for the flow, or update existing entry."""
try:
youtube = await self.get_resource(data[CONF_TOKEN][CONF_ACCESS_TOKEN])
own_channel = await first(youtube.get_user_channels())
if own_channel is None or own_channel.snippet is None:
own_channels = [
channel
async for channel in youtube.get_user_channels()
if channel.snippet is not None
]
if not own_channels:
return self.async_abort(
reason="no_channel",
description_placeholders={"support_url": CHANNEL_CREATION_HELP_URL},
@ -111,10 +114,10 @@ class OAuth2FlowHandler(
except Exception as ex: # noqa: BLE001
LOGGER.error("Unknown error occurred: %s", ex.args)
return self.async_abort(reason="unknown")
self._title = own_channel.snippet.title
self._title = own_channels[0].snippet.title
self._data = data
await self.async_set_unique_id(own_channel.channel_id)
await self.async_set_unique_id(own_channels[0].channel_id)
if self.source != SOURCE_REAUTH:
self._abort_if_unique_id_configured()
@ -138,13 +141,39 @@ class OAuth2FlowHandler(
options=user_input,
)
youtube = await self.get_resource(self._data[CONF_TOKEN][CONF_ACCESS_TOKEN])
# Get user's own channels
own_channels = [
channel
async for channel in youtube.get_user_channels()
if channel.snippet is not None
]
if not own_channels:
return self.async_abort(
reason="no_channel",
description_placeholders={"support_url": CHANNEL_CREATION_HELP_URL},
)
# Start with user's own channels
selectable_channels = [
SelectOptionDict(
value=subscription.snippet.channel_id,
label=subscription.snippet.title,
value=channel.channel_id,
label=f"{channel.snippet.title} (Your Channel)",
)
async for subscription in youtube.get_user_subscriptions()
for channel in own_channels
]
# Add subscribed channels
selectable_channels.extend(
[
SelectOptionDict(
value=subscription.snippet.channel_id,
label=subscription.snippet.title,
)
async for subscription in youtube.get_user_subscriptions()
]
)
if not selectable_channels:
return self.async_abort(reason="no_subscriptions")
return self.async_show_form(
@ -175,13 +204,39 @@ class YouTubeOptionsFlowHandler(OptionsFlow):
await youtube.set_user_authentication(
self.config_entry.data[CONF_TOKEN][CONF_ACCESS_TOKEN], [AuthScope.READ_ONLY]
)
# Get user's own channels
own_channels = [
channel
async for channel in youtube.get_user_channels()
if channel.snippet is not None
]
if not own_channels:
return self.async_abort(
reason="no_channel",
description_placeholders={"support_url": CHANNEL_CREATION_HELP_URL},
)
# Start with user's own channels
selectable_channels = [
SelectOptionDict(
value=subscription.snippet.channel_id,
label=subscription.snippet.title,
value=channel.channel_id,
label=f"{channel.snippet.title} (Your Channel)",
)
async for subscription in youtube.get_user_subscriptions()
for channel in own_channels
]
# Add subscribed channels
selectable_channels.extend(
[
SelectOptionDict(
value=subscription.snippet.channel_id,
label=subscription.snippet.title,
)
async for subscription in youtube.get_user_subscriptions()
]
)
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(

181
tests/components/youtube/test_config_flow.py
View File

@ -131,7 +131,51 @@ async def test_flow_abort_without_subscriptions(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
) -> None:
"""Check abort flow if user has no subscriptions."""
"""Check abort flow if user has no subscriptions and no own channel."""
result = await hass.config_entries.flow.async_init(
"youtube", context={"source": config_entries.SOURCE_USER}
)
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"{GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope={'+'.join(SCOPES)}"
"&access_type=offline&prompt=consent"
)
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"
service = MockYouTube(
channel_fixture="youtube/get_no_channel.json",
subscriptions_fixture="youtube/get_no_subscriptions.json",
)
with (
patch("homeassistant.components.youtube.async_setup_entry", return_value=True),
patch(
"homeassistant.components.youtube.config_flow.YouTube", return_value=service
),
):
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_channel"
@pytest.mark.usefixtures("current_request_with_host")
async def test_flow_without_subscriptions(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
) -> None:
"""Check flow continues even without subscriptions since user has their own channel."""
result = await hass.config_entries.flow.async_init(
"youtube", context={"source": config_entries.SOURCE_USER}
)
@ -163,8 +207,30 @@ async def test_flow_abort_without_subscriptions(
),
):
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_subscriptions"
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "channels"
# Verify the form schema contains only the user's own channel
schema = result["data_schema"]
channels = schema.schema[CONF_CHANNELS].config["options"]
assert len(channels) == 1
assert channels[0]["value"] == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
assert "(Your Channel)" in channels[0]["label"]
# Test selecting the own channel
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TITLE
assert "result" in result
assert result["result"].unique_id == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
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["options"] == {CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]}
@pytest.mark.usefixtures("current_request_with_host")
@ -373,3 +439,112 @@ async def test_options_flow(
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]}
@pytest.mark.usefixtures("current_request_with_host")
async def test_own_channel_included(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
) -> None:
"""Test that the user's own channel is included in the list of selectable channels."""
result = await hass.config_entries.flow.async_init(
"youtube", context={"source": config_entries.SOURCE_USER}
)
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"{GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope={'+'.join(SCOPES)}"
"&access_type=offline&prompt=consent"
)
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"
with (
patch(
"homeassistant.components.youtube.async_setup_entry", return_value=True
) as mock_setup,
patch(
"homeassistant.components.youtube.config_flow.YouTube",
return_value=MockYouTube(),
),
):
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "channels"
# Verify the form schema contains the user's own channel
schema = result["data_schema"]
channels = schema.schema[CONF_CHANNELS].config["options"]
assert any(
channel["value"] == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
and "(Your Channel)" in channel["label"]
for channel in channels
)
# Test selecting both own channel and a subscribed channel
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw", "UC_x5XG1OV2P6uZZ5FSM9Ttw"]
},
)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TITLE
assert "result" in result
assert result["result"].unique_id == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
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["options"] == {
CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw", "UC_x5XG1OV2P6uZZ5FSM9Ttw"]
}
async def test_options_flow_own_channel(
hass: HomeAssistant, setup_integration: ComponentSetup
) -> None:
"""Test the options flow includes the user's own channel."""
await setup_integration()
with patch(
"homeassistant.components.youtube.config_flow.YouTube",
return_value=MockYouTube(),
):
entry = hass.config_entries.async_entries(DOMAIN)[0]
result = await hass.config_entries.options.async_init(entry.entry_id)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
# Verify the form schema contains the user's own channel
schema = result["data_schema"]
channels = schema.schema[CONF_CHANNELS].config["options"]
assert any(
channel["value"] == "UC_x5XG1OV2P6uZZ5FSM9Ttw"
and "(Your Channel)" in channel["label"]
for channel in channels
)
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {CONF_CHANNELS: ["UC_x5XG1OV2P6uZZ5FSM9Ttw"]}