mirror of
https://github.com/home-assistant/core.git
synced 2025-07-21 04:07:08 +00:00
Lastfm config flow (#92299)
* Move constant values to separate file * Move constant values to separate file * Add config flow to lastfm * Add tests * Add config flow to lastfm * Add tests * Add tests * Add tests * Add extra form for main user and autofill with friends * Add extra form for main user and autofill with friends * Add extra form for main user and autofill with friends * Add extra form for main user and autofill with friends * Add OptionsFlow * Add tests * Fix feedback * Fix feedback * Fix feedback * Fix feedback * Fix test * Apply suggestions from code review Co-authored-by: G Johansson <goran.johansson@shiftit.se> * Update config_flow.py * Update config_flow.py * Update config_flow.py * Update homeassistant/components/lastfm/config_flow.py Co-authored-by: G Johansson <goran.johansson@shiftit.se> * Add tests * Cleanup * Update config_flow.py * Update config_flow.py * Update config_flow.py * Fix test * Fix feedback * Codeowner lastfm * Fix feedback * Fix feedback * Parametrize errors * Parametrize errors * Parametrize errors * Finish tests --------- Co-authored-by: G Johansson <goran.johansson@shiftit.se>
This commit is contained in:
Joost Lekkerkerker
GitHub
parent
e09e4f14d6
commit
a96215bf2e
@ -651,6 +651,8 @@ build.json @home-assistant/supervisor
|
||||
/tests/components/lametric/ @robbiet480 @frenck @bachya
|
||||
/homeassistant/components/landisgyr_heat_meter/ @vpathuis
|
||||
/tests/components/landisgyr_heat_meter/ @vpathuis
|
||||
/homeassistant/components/lastfm/ @joostlek
|
||||
/tests/components/lastfm/ @joostlek
|
||||
/homeassistant/components/launch_library/ @ludeeus @DurgNomis-drol
|
||||
/tests/components/launch_library/ @ludeeus @DurgNomis-drol
|
||||
/homeassistant/components/laundrify/ @xLarry
|
||||
|
||||
@ -1 +1,27 @@
|
||||
"""The lastfm component."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import PLATFORMS
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up lastfm from a config entry."""
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
entry.async_on_unload(entry.add_update_listener(update_listener))
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload lastfm config entry."""
|
||||
|
||||
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
||||
|
||||
async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
|
||||
"""Handle options update."""
|
||||
await hass.config_entries.async_reload(entry.entry_id)
|
||||
|
||||
223
homeassistant/components/lastfm/config_flow.py
Normal file
223
homeassistant/components/lastfm/config_flow.py
Normal file
@ -0,0 +1,223 @@
|
||||
"""Config flow for LastFm."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from pylast import LastFMNetwork, User, WSError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import (
|
||||
ConfigEntry,
|
||||
ConfigFlow,
|
||||
OptionsFlowWithConfigEntry,
|
||||
)
|
||||
from homeassistant.const import CONF_API_KEY
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.helpers.selector import (
|
||||
SelectOptionDict,
|
||||
SelectSelector,
|
||||
SelectSelectorConfig,
|
||||
)
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from .const import CONF_MAIN_USER, CONF_USERS, DOMAIN
|
||||
|
||||
PLACEHOLDERS = {"api_account_url": "https://www.last.fm/api/account/create"}
|
||||
|
||||
CONFIG_SCHEMA: vol.Schema = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_API_KEY): str,
|
||||
vol.Required(CONF_MAIN_USER): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def get_lastfm_user(api_key: str, username: str) -> tuple[User, dict[str, str]]:
|
||||
"""Get and validate lastFM User."""
|
||||
user = LastFMNetwork(api_key=api_key).get_user(username)
|
||||
errors = {}
|
||||
try:
|
||||
user.get_playcount()
|
||||
except WSError as error:
|
||||
if error.details == "User not found":
|
||||
errors["base"] = "invalid_account"
|
||||
elif (
|
||||
error.details
|
||||
== "Invalid API key - You must be granted a valid key by last.fm"
|
||||
):
|
||||
errors["base"] = "invalid_auth"
|
||||
else:
|
||||
errors["base"] = "unknown"
|
||||
except Exception: # pylint:disable=broad-except
|
||||
errors["base"] = "unknown"
|
||||
return user, errors
|
||||
|
||||
|
||||
def validate_lastfm_users(
|
||||
api_key: str, usernames: list[str]
|
||||
) -> tuple[list[str], dict[str, str]]:
|
||||
"""Validate list of users. Return tuple of valid users and errors."""
|
||||
valid_users = []
|
||||
errors = {}
|
||||
for username in usernames:
|
||||
_, lastfm_errors = get_lastfm_user(api_key, username)
|
||||
if lastfm_errors:
|
||||
errors = lastfm_errors
|
||||
else:
|
||||
valid_users.append(username)
|
||||
return valid_users, errors
|
||||
|
||||
|
||||
class LastFmConfigFlowHandler(ConfigFlow, domain=DOMAIN):
|
||||
"""Config flow handler for LastFm."""
|
||||
|
||||
data: dict[str, Any] = {}
|
||||
|
||||
@staticmethod
|
||||
@callback
|
||||
def async_get_options_flow(
|
||||
config_entry: ConfigEntry,
|
||||
) -> LastFmOptionsFlowHandler:
|
||||
"""Get the options flow for this handler."""
|
||||
return LastFmOptionsFlowHandler(config_entry)
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Initialize user input."""
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
self.data = user_input.copy()
|
||||
_, errors = get_lastfm_user(
|
||||
self.data[CONF_API_KEY], self.data[CONF_MAIN_USER]
|
||||
)
|
||||
if not errors:
|
||||
return await self.async_step_friends()
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
errors=errors,
|
||||
description_placeholders=PLACEHOLDERS,
|
||||
data_schema=self.add_suggested_values_to_schema(CONFIG_SCHEMA, user_input),
|
||||
)
|
||||
|
||||
async def async_step_friends(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Form to select other users and friends."""
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
users, errors = validate_lastfm_users(
|
||||
self.data[CONF_API_KEY], user_input[CONF_USERS]
|
||||
)
|
||||
user_input[CONF_USERS] = users
|
||||
if not errors:
|
||||
return self.async_create_entry(
|
||||
title="LastFM",
|
||||
data={},
|
||||
options={
|
||||
CONF_API_KEY: self.data[CONF_API_KEY],
|
||||
CONF_MAIN_USER: self.data[CONF_MAIN_USER],
|
||||
CONF_USERS: [
|
||||
self.data[CONF_MAIN_USER],
|
||||
*user_input[CONF_USERS],
|
||||
],
|
||||
},
|
||||
)
|
||||
try:
|
||||
main_user, _ = get_lastfm_user(
|
||||
self.data[CONF_API_KEY], self.data[CONF_MAIN_USER]
|
||||
)
|
||||
friends = [
|
||||
SelectOptionDict(value=friend.name, label=friend.get_name(True))
|
||||
for friend in main_user.get_friends()
|
||||
]
|
||||
except WSError:
|
||||
friends = []
|
||||
return self.async_show_form(
|
||||
step_id="friends",
|
||||
errors=errors,
|
||||
data_schema=self.add_suggested_values_to_schema(
|
||||
vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_USERS): SelectSelector(
|
||||
SelectSelectorConfig(
|
||||
options=friends, custom_value=True, multiple=True
|
||||
)
|
||||
),
|
||||
}
|
||||
),
|
||||
user_input or {CONF_USERS: []},
|
||||
),
|
||||
)
|
||||
|
||||
async def async_step_import(self, import_config: ConfigType) -> FlowResult:
|
||||
"""Import config from yaml."""
|
||||
for entry in self._async_current_entries():
|
||||
if entry.options[CONF_API_KEY] == import_config[CONF_API_KEY]:
|
||||
return self.async_abort(reason="already_configured")
|
||||
users, _ = validate_lastfm_users(
|
||||
import_config[CONF_API_KEY], import_config[CONF_USERS]
|
||||
)
|
||||
return self.async_create_entry(
|
||||
title="LastFM",
|
||||
data={},
|
||||
options={
|
||||
CONF_API_KEY: import_config[CONF_API_KEY],
|
||||
CONF_MAIN_USER: None,
|
||||
CONF_USERS: users,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
class LastFmOptionsFlowHandler(OptionsFlowWithConfigEntry):
|
||||
"""LastFm Options flow handler."""
|
||||
|
||||
async def async_step_init(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Initialize form."""
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
users, errors = validate_lastfm_users(
|
||||
self.options[CONF_API_KEY], user_input[CONF_USERS]
|
||||
)
|
||||
user_input[CONF_USERS] = users
|
||||
if not errors:
|
||||
return self.async_create_entry(
|
||||
title="LastFM",
|
||||
data={
|
||||
**self.options,
|
||||
CONF_USERS: user_input[CONF_USERS],
|
||||
},
|
||||
)
|
||||
if self.options[CONF_MAIN_USER]:
|
||||
try:
|
||||
main_user, _ = get_lastfm_user(
|
||||
self.options[CONF_API_KEY],
|
||||
self.options[CONF_MAIN_USER],
|
||||
)
|
||||
friends = [
|
||||
SelectOptionDict(value=friend.name, label=friend.get_name(True))
|
||||
for friend in main_user.get_friends()
|
||||
]
|
||||
except WSError:
|
||||
friends = []
|
||||
else:
|
||||
friends = []
|
||||
return self.async_show_form(
|
||||
step_id="init",
|
||||
errors=errors,
|
||||
data_schema=self.add_suggested_values_to_schema(
|
||||
vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_USERS): SelectSelector(
|
||||
SelectSelectorConfig(
|
||||
options=friends, custom_value=True, multiple=True
|
||||
)
|
||||
),
|
||||
}
|
||||
),
|
||||
user_input or self.options,
|
||||
),
|
||||
)
|
||||
@ -2,10 +2,14 @@
|
||||
import logging
|
||||
from typing import Final
|
||||
|
||||
from homeassistant.const import Platform
|
||||
|
||||
LOGGER = logging.getLogger(__package__)
|
||||
DOMAIN: Final = "lastfm"
|
||||
PLATFORMS = [Platform.SENSOR]
|
||||
DEFAULT_NAME = "LastFM"
|
||||
|
||||
CONF_MAIN_USER = "main_user"
|
||||
CONF_USERS = "users"
|
||||
|
||||
ATTR_LAST_PLAYED = "last_played"
|
||||
|
||||
@ -1,7 +1,8 @@
|
||||
{
|
||||
"domain": "lastfm",
|
||||
"name": "Last.fm",
|
||||
"codeowners": [],
|
||||
"codeowners": ["@joostlek"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/lastfm",
|
||||
"iot_class": "cloud_polling",
|
||||
"loggers": ["pylast"],
|
||||
|
||||
@ -7,10 +7,14 @@ from pylast import LastFMNetwork, Track, User, WSError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import CONF_API_KEY
|
||||
from homeassistant.core import HomeAssistant
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.device_registry import DeviceEntryType
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
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 (
|
||||
@ -18,6 +22,8 @@ from .const import (
|
||||
ATTR_PLAY_COUNT,
|
||||
ATTR_TOP_PLAYED,
|
||||
CONF_USERS,
|
||||
DEFAULT_NAME,
|
||||
DOMAIN,
|
||||
LOGGER,
|
||||
STATE_NOT_SCROBBLING,
|
||||
)
|
||||
@ -35,23 +41,46 @@ def format_track(track: Track) -> str:
|
||||
return f"{track.artist} - {track.title}"
|
||||
|
||||
|
||||
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 Last.fm sensor platform."""
|
||||
lastfm_api = LastFMNetwork(api_key=config[CONF_API_KEY])
|
||||
entities = []
|
||||
for username in config[CONF_USERS]:
|
||||
try:
|
||||
user = lastfm_api.get_user(username)
|
||||
entities.append(LastFmSensor(user, lastfm_api))
|
||||
except WSError as exc:
|
||||
LOGGER.error("Failed to load LastFM user `%s`: %r", username, exc)
|
||||
return
|
||||
add_entities(entities, True)
|
||||
"""Set up the Last.fm sensor platform from yaml."""
|
||||
|
||||
async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
"deprecated_yaml",
|
||||
breaks_in_ha_version="2023.8.0",
|
||||
is_fixable=False,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml",
|
||||
)
|
||||
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Initialize the entries."""
|
||||
|
||||
lastfm_api = LastFMNetwork(api_key=entry.options[CONF_API_KEY])
|
||||
async_add_entities(
|
||||
(
|
||||
LastFmSensor(lastfm_api.get_user(user), entry.entry_id)
|
||||
for user in entry.options[CONF_USERS]
|
||||
),
|
||||
True,
|
||||
)
|
||||
|
||||
|
||||
class LastFmSensor(SensorEntity):
|
||||
@ -60,14 +89,27 @@ class LastFmSensor(SensorEntity):
|
||||
_attr_attribution = "Data provided by Last.fm"
|
||||
_attr_icon = "mdi:radio-fm"
|
||||
|
||||
def __init__(self, user: User, lastfm_api: LastFMNetwork) -> None:
|
||||
def __init__(self, user: User, entry_id: str) -> None:
|
||||
"""Initialize the sensor."""
|
||||
self._user = user
|
||||
self._attr_unique_id = hashlib.sha256(user.name.encode("utf-8")).hexdigest()
|
||||
self._attr_name = user.name
|
||||
self._user = user
|
||||
self._attr_device_info = DeviceInfo(
|
||||
configuration_url="https://www.last.fm",
|
||||
entry_type=DeviceEntryType.SERVICE,
|
||||
identifiers={(DOMAIN, f"{entry_id}_{self._attr_unique_id}")},
|
||||
manufacturer=DEFAULT_NAME,
|
||||
name=f"{DEFAULT_NAME} {user.name}",
|
||||
)
|
||||
|
||||
def update(self) -> None:
|
||||
"""Update device state."""
|
||||
try:
|
||||
self._user.get_playcount()
|
||||
except WSError as exc:
|
||||
self._attr_available = False
|
||||
LOGGER.error("Failed to load LastFM user `%s`: %r", self._user.name, exc)
|
||||
return
|
||||
self._attr_entity_picture = self._user.get_image()
|
||||
if now_playing := self._user.get_now_playing():
|
||||
self._attr_native_value = format_track(now_playing)
|
||||
|
||||
45
homeassistant/components/lastfm/strings.json
Normal file
45
homeassistant/components/lastfm/strings.json
Normal file
@ -0,0 +1,45 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"description": "Request an API account at {api_account_url}.",
|
||||
"data": {
|
||||
"api_key": "[%key:common::config_flow::data::api_key%]",
|
||||
"main_user": "Last.fm username"
|
||||
}
|
||||
},
|
||||
"friends": {
|
||||
"description": "Fill in other users you want to add.",
|
||||
"data": {
|
||||
"users": "Last.fm usernames"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"invalid_account": "Invalid username",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"description": "Fill in other users you want to add.",
|
||||
"data": {
|
||||
"users": "Last.fm usernames"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"invalid_account": "Invalid username",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
"deprecated_yaml": {
|
||||
"title": "The LastFM YAML configuration is being removed",
|
||||
"description": "Configuring LastFM using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the LastFM YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -234,6 +234,7 @@ FLOWS = {
|
||||
"lacrosse_view",
|
||||
"lametric",
|
||||
"landisgyr_heat_meter",
|
||||
"lastfm",
|
||||
"launch_library",
|
||||
"laundrify",
|
||||
"ld2410_ble",
|
||||
|
||||
@ -2846,7 +2846,7 @@
|
||||
"lastfm": {
|
||||
"name": "Last.fm",
|
||||
"integration_type": "hub",
|
||||
"config_flow": false,
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_polling"
|
||||
},
|
||||
"launch_library": {
|
||||
|
||||
@ -1 +1,87 @@
|
||||
"""The tests for lastfm."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from pylast import Track, WSError
|
||||
|
||||
from homeassistant.components.lastfm.const import CONF_MAIN_USER, CONF_USERS
|
||||
from homeassistant.const import CONF_API_KEY
|
||||
|
||||
API_KEY = "asdasdasdasdasd"
|
||||
USERNAME_1 = "testaccount1"
|
||||
USERNAME_2 = "testaccount2"
|
||||
|
||||
CONF_DATA = {
|
||||
CONF_API_KEY: API_KEY,
|
||||
CONF_MAIN_USER: USERNAME_1,
|
||||
CONF_USERS: [USERNAME_1, USERNAME_2],
|
||||
}
|
||||
CONF_USER_DATA = {CONF_API_KEY: API_KEY, CONF_MAIN_USER: USERNAME_1}
|
||||
CONF_FRIENDS_DATA = {CONF_USERS: [USERNAME_2]}
|
||||
|
||||
|
||||
class MockNetwork:
|
||||
"""Mock _Network object for pylast."""
|
||||
|
||||
def __init__(self, username: str):
|
||||
"""Initialize the mock."""
|
||||
self.username = username
|
||||
|
||||
|
||||
class MockUser:
|
||||
"""Mock User object for pylast."""
|
||||
|
||||
def __init__(self, now_playing_result, error, has_friends, username):
|
||||
"""Initialize the mock."""
|
||||
self._now_playing_result = now_playing_result
|
||||
self._thrown_error = error
|
||||
self._has_friends = has_friends
|
||||
self.name = username
|
||||
|
||||
def get_name(self, capitalized: bool) -> str:
|
||||
"""Get name of the user."""
|
||||
return self.name
|
||||
|
||||
def get_playcount(self):
|
||||
"""Get mock play count."""
|
||||
if self._thrown_error:
|
||||
raise self._thrown_error
|
||||
return 1
|
||||
|
||||
def get_image(self):
|
||||
"""Get mock image."""
|
||||
|
||||
def get_recent_tracks(self, limit):
|
||||
"""Get mock recent tracks."""
|
||||
return []
|
||||
|
||||
def get_top_tracks(self, limit):
|
||||
"""Get mock top tracks."""
|
||||
return []
|
||||
|
||||
def get_now_playing(self):
|
||||
"""Get mock now playing."""
|
||||
return self._now_playing_result
|
||||
|
||||
def get_friends(self):
|
||||
"""Get mock friends."""
|
||||
if self._has_friends is False:
|
||||
raise WSError("network", "status", "Page not found")
|
||||
return [MockUser(None, None, True, USERNAME_2)]
|
||||
|
||||
|
||||
def patch_fetch_user(
|
||||
now_playing: Track | None = None,
|
||||
thrown_error: Exception | None = None,
|
||||
has_friends: bool = True,
|
||||
username: str = USERNAME_1,
|
||||
) -> MockUser:
|
||||
"""Patch interface."""
|
||||
return patch(
|
||||
"pylast.User",
|
||||
return_value=MockUser(now_playing, thrown_error, has_friends, username),
|
||||
)
|
||||
|
||||
|
||||
def patch_setup_entry() -> bool:
|
||||
"""Patch interface."""
|
||||
return patch("homeassistant.components.lastfm.async_setup_entry", return_value=True)
|
||||
|
||||
305
tests/components/lastfm/test_config_flow.py
Normal file
305
tests/components/lastfm/test_config_flow.py
Normal file
@ -0,0 +1,305 @@
|
||||
"""Test Lastfm config flow."""
|
||||
from pylast import WSError
|
||||
import pytest
|
||||
|
||||
from homeassistant import data_entry_flow
|
||||
from homeassistant.components.lastfm.const import (
|
||||
CONF_MAIN_USER,
|
||||
CONF_USERS,
|
||||
DEFAULT_NAME,
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
|
||||
from homeassistant.const import CONF_API_KEY
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from . import (
|
||||
API_KEY,
|
||||
CONF_DATA,
|
||||
CONF_FRIENDS_DATA,
|
||||
CONF_USER_DATA,
|
||||
USERNAME_1,
|
||||
USERNAME_2,
|
||||
patch_fetch_user,
|
||||
patch_setup_entry,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_full_user_flow(hass: HomeAssistant) -> None:
|
||||
"""Test the full user configuration flow."""
|
||||
with patch_fetch_user(), patch_setup_entry():
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=CONF_USER_DATA,
|
||||
)
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert not result["errors"]
|
||||
assert result["step_id"] == "friends"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=CONF_FRIENDS_DATA
|
||||
)
|
||||
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == DEFAULT_NAME
|
||||
assert result["options"] == CONF_DATA
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("error", "message"),
|
||||
[
|
||||
(
|
||||
WSError(
|
||||
"network",
|
||||
"status",
|
||||
"Invalid API key - You must be granted a valid key by last.fm",
|
||||
),
|
||||
"invalid_auth",
|
||||
),
|
||||
(WSError("network", "status", "User not found"), "invalid_account"),
|
||||
(Exception(), "unknown"),
|
||||
(WSError("network", "status", "Something strange"), "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_flow_fails(hass: HomeAssistant, error: Exception, message: str) -> None:
|
||||
"""Test user initialized flow with invalid username."""
|
||||
with patch_fetch_user(thrown_error=error):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}, data=CONF_USER_DATA
|
||||
)
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"]["base"] == message
|
||||
|
||||
with patch_fetch_user(), patch_setup_entry():
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=CONF_USER_DATA,
|
||||
)
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert not result["errors"]
|
||||
assert result["step_id"] == "friends"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=CONF_FRIENDS_DATA
|
||||
)
|
||||
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == DEFAULT_NAME
|
||||
assert result["options"] == CONF_DATA
|
||||
|
||||
|
||||
async def test_flow_friends_invalid_username(hass: HomeAssistant) -> None:
|
||||
"""Test user initialized flow with invalid username."""
|
||||
with patch_fetch_user(), patch_setup_entry():
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=CONF_USER_DATA,
|
||||
)
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "friends"
|
||||
|
||||
with patch_fetch_user(thrown_error=WSError("network", "status", "User not found")):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=CONF_FRIENDS_DATA
|
||||
)
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "friends"
|
||||
assert result["errors"]["base"] == "invalid_account"
|
||||
|
||||
with patch_fetch_user(), patch_setup_entry():
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=CONF_FRIENDS_DATA
|
||||
)
|
||||
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == DEFAULT_NAME
|
||||
assert result["options"] == CONF_DATA
|
||||
|
||||
|
||||
async def test_flow_friends_no_friends(hass: HomeAssistant) -> None:
|
||||
"""Test options is empty when user has no friends."""
|
||||
with patch_fetch_user(has_friends=False), patch_setup_entry():
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=CONF_USER_DATA,
|
||||
)
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "friends"
|
||||
assert len(result["data_schema"].schema[CONF_USERS].config["options"]) == 0
|
||||
|
||||
|
||||
async def test_import_flow_success(hass: HomeAssistant) -> None:
|
||||
"""Test import flow."""
|
||||
with patch_fetch_user():
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={CONF_API_KEY: API_KEY, CONF_USERS: [USERNAME_1, USERNAME_2]},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "LastFM"
|
||||
assert result["options"] == {
|
||||
"api_key": "asdasdasdasdasd",
|
||||
"main_user": None,
|
||||
"users": ["testaccount1", "testaccount2"],
|
||||
}
|
||||
|
||||
|
||||
async def test_import_flow_already_exist(hass: HomeAssistant) -> None:
|
||||
"""Test import of yaml already exist."""
|
||||
|
||||
MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={CONF_API_KEY: API_KEY, CONF_USERS: ["test"]},
|
||||
).add_to_hass(hass)
|
||||
|
||||
with patch_fetch_user():
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=CONF_DATA,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_options_flow(hass: HomeAssistant) -> None:
|
||||
"""Test updating options."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={
|
||||
CONF_API_KEY: API_KEY,
|
||||
CONF_MAIN_USER: USERNAME_1,
|
||||
CONF_USERS: [USERNAME_1, USERNAME_2],
|
||||
},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
with patch_fetch_user():
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_USERS: [USERNAME_1]},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
|
||||
assert result["data"] == {
|
||||
CONF_API_KEY: API_KEY,
|
||||
CONF_MAIN_USER: USERNAME_1,
|
||||
CONF_USERS: [USERNAME_1],
|
||||
}
|
||||
|
||||
|
||||
async def test_options_flow_incorrect_username(hass: HomeAssistant) -> None:
|
||||
"""Test updating options doesn't work with incorrect username."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={
|
||||
CONF_API_KEY: API_KEY,
|
||||
CONF_MAIN_USER: USERNAME_1,
|
||||
CONF_USERS: [USERNAME_1],
|
||||
},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
with patch_fetch_user():
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
with patch_fetch_user(thrown_error=WSError("network", "status", "User not found")):
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_USERS: [USERNAME_1]},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "init"
|
||||
assert result["errors"]["base"] == "invalid_account"
|
||||
|
||||
with patch_fetch_user():
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_USERS: [USERNAME_1]},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
|
||||
assert result["data"] == {
|
||||
CONF_API_KEY: API_KEY,
|
||||
CONF_MAIN_USER: USERNAME_1,
|
||||
CONF_USERS: [USERNAME_1],
|
||||
}
|
||||
|
||||
|
||||
async def test_options_flow_from_import(hass: HomeAssistant) -> None:
|
||||
"""Test updating options gained from import."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={
|
||||
CONF_API_KEY: API_KEY,
|
||||
CONF_MAIN_USER: None,
|
||||
CONF_USERS: [USERNAME_1],
|
||||
},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
with patch_fetch_user():
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "init"
|
||||
assert len(result["data_schema"].schema[CONF_USERS].config["options"]) == 0
|
||||
|
||||
|
||||
async def test_options_flow_without_friends(hass: HomeAssistant) -> None:
|
||||
"""Test updating options for someone without friends."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={
|
||||
CONF_API_KEY: API_KEY,
|
||||
CONF_MAIN_USER: USERNAME_1,
|
||||
CONF_USERS: [USERNAME_1],
|
||||
},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
with patch_fetch_user(has_friends=False):
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
assert result["step_id"] == "init"
|
||||
assert len(result["data_schema"].schema[CONF_USERS].config["options"]) == 0
|
||||
36
tests/components/lastfm/test_init.py
Normal file
36
tests/components/lastfm/test_init.py
Normal file
@ -0,0 +1,36 @@
|
||||
"""Test LastFM component setup process."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.lastfm.const import CONF_MAIN_USER, CONF_USERS, DOMAIN
|
||||
from homeassistant.const import CONF_API_KEY
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import USERNAME_1, USERNAME_2, patch_fetch_user
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_load_unload_entry(hass: HomeAssistant) -> None:
|
||||
"""Test load and unload entry."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={
|
||||
CONF_API_KEY: "12345678",
|
||||
CONF_MAIN_USER: [USERNAME_1],
|
||||
CONF_USERS: [USERNAME_1, USERNAME_2],
|
||||
},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
with patch_fetch_user():
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("sensor.testaccount1")
|
||||
assert state
|
||||
|
||||
await hass.config_entries.async_remove(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("sensor.testaccount1")
|
||||
assert not state
|
||||
@ -1,94 +1,37 @@
|
||||
"""Tests for the lastfm sensor."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from pylast import Track
|
||||
import pytest
|
||||
|
||||
from homeassistant.components import sensor
|
||||
from homeassistant.components.lastfm.const import STATE_NOT_SCROBBLING
|
||||
from homeassistant.components.lastfm.const import DOMAIN, STATE_NOT_SCROBBLING
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import CONF_DATA, MockNetwork, patch_fetch_user
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
class MockNetwork:
|
||||
"""Mock _Network object for pylast."""
|
||||
|
||||
def __init__(self, username: str):
|
||||
"""Initialize the mock."""
|
||||
self.username = username
|
||||
|
||||
|
||||
class MockUser:
|
||||
"""Mock User object for pylast."""
|
||||
|
||||
def __init__(self, now_playing_result):
|
||||
"""Initialize the mock."""
|
||||
self._now_playing_result = now_playing_result
|
||||
self.name = "test"
|
||||
|
||||
def get_playcount(self):
|
||||
"""Get mock play count."""
|
||||
return 1
|
||||
|
||||
def get_image(self):
|
||||
"""Get mock image."""
|
||||
|
||||
def get_recent_tracks(self, limit):
|
||||
"""Get mock recent tracks."""
|
||||
return []
|
||||
|
||||
def get_top_tracks(self, limit):
|
||||
"""Get mock top tracks."""
|
||||
return []
|
||||
|
||||
def get_now_playing(self):
|
||||
"""Get mock now playing."""
|
||||
return self._now_playing_result
|
||||
|
||||
|
||||
@pytest.fixture(name="lastfm_network")
|
||||
def lastfm_network_fixture():
|
||||
"""Create fixture for LastFMNetwork."""
|
||||
with patch(
|
||||
"homeassistant.components.lastfm.sensor.LastFMNetwork"
|
||||
) as lastfm_network:
|
||||
yield lastfm_network
|
||||
|
||||
|
||||
async def test_update_not_playing(hass: HomeAssistant, lastfm_network) -> None:
|
||||
async def test_update_not_playing(hass: HomeAssistant) -> None:
|
||||
"""Test update when no playing song."""
|
||||
|
||||
lastfm_network.return_value.get_user.return_value = MockUser(None)
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
sensor.DOMAIN,
|
||||
{"sensor": {"platform": "lastfm", "api_key": "secret-key", "users": ["test"]}},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity_id = "sensor.test"
|
||||
entry = MockConfigEntry(domain=DOMAIN, data={}, options=CONF_DATA)
|
||||
entry.add_to_hass(hass)
|
||||
with patch_fetch_user(None):
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
entity_id = "sensor.testaccount1"
|
||||
|
||||
state = hass.states.get(entity_id)
|
||||
|
||||
assert state.state == STATE_NOT_SCROBBLING
|
||||
|
||||
|
||||
async def test_update_playing(hass: HomeAssistant, lastfm_network) -> None:
|
||||
"""Test update when song playing."""
|
||||
|
||||
lastfm_network.return_value.get_user.return_value = MockUser(
|
||||
Track("artist", "title", MockNetwork("test"))
|
||||
)
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
sensor.DOMAIN,
|
||||
{"sensor": {"platform": "lastfm", "api_key": "secret-key", "users": ["test"]}},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity_id = "sensor.test"
|
||||
async def test_update_playing(hass: HomeAssistant) -> None:
|
||||
"""Test update when playing a song."""
|
||||
entry = MockConfigEntry(domain=DOMAIN, data={}, options=CONF_DATA)
|
||||
entry.add_to_hass(hass)
|
||||
with patch_fetch_user(Track("artist", "title", MockNetwork("test"))):
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
entity_id = "sensor.testaccount1"
|
||||
|
||||
state = hass.states.get(entity_id)
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user