home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add config flow to steam_online integration (#67261)

Browse Source 
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
Robert Hillis
2022-04-27 02:07:21 -04:00
committed by GitHub
parent 5f63944142
commit b1a6521abd
14 changed files with 931 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS
View File

@ -965,6 +965,8 @@ build.json @home-assistant/supervisor
/tests/components/starline/ @anonym-tsk
/homeassistant/components/statistics/ @fabaff @ThomDietrich
/tests/components/statistics/ @fabaff @ThomDietrich
/homeassistant/components/steam_online/ @tkdrob
/tests/components/steam_online/ @tkdrob
/homeassistant/components/steamist/ @bdraco
/tests/components/steamist/ @bdraco
/homeassistant/components/stiebel_eltron/ @fucm

49
homeassistant/components/steam_online/__init__.py
View File

@ -1 +1,48 @@
"""The steam_online component."""
"""The Steam integration."""
from __future__ import annotations
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DEFAULT_NAME, DOMAIN
from .coordinator import SteamDataUpdateCoordinator
PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Steam from a config entry."""
coordinator = SteamDataUpdateCoordinator(hass)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
class SteamEntity(CoordinatorEntity[SteamDataUpdateCoordinator]):
"""Representation of a Steam entity."""
_attr_attribution = "Data provided by Steam"
def __init__(self, coordinator: SteamDataUpdateCoordinator) -> None:
"""Initialize a Steam entity."""
super().__init__(coordinator)
self._attr_device_info = DeviceInfo(
configuration_url="https://store.steampowered.com",
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, coordinator.config_entry.entry_id)},
manufacturer=DEFAULT_NAME,
name=DEFAULT_NAME,
)

211
homeassistant/components/steam_online/config_flow.py Normal file
View File

@ -0,0 +1,211 @@
"""Config flow for Steam integration."""
from __future__ import annotations
from typing import Any
import steam
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY, Platform
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.typing import ConfigType
from .const import CONF_ACCOUNT, CONF_ACCOUNTS, DEFAULT_NAME, DOMAIN, LOGGER
def validate_input(user_input: dict[str, str | int]) -> list[dict[str, str | int]]:
"""Handle common flow input validation."""
steam.api.key.set(user_input[CONF_API_KEY])
interface = steam.api.interface("ISteamUser")
names = interface.GetPlayerSummaries(steamids=user_input[CONF_ACCOUNT])
return names["response"]["players"]["player"]
class SteamFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Steam."""
def __init__(self) -> None:
"""Initialize the flow."""
self.entry: config_entries.ConfigEntry | None = None
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> config_entries.OptionsFlow:
"""Get the options flow for this handler."""
return SteamOptionsFlowHandler(config_entry)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
errors = {}
if user_input is None and self.entry:
user_input = {CONF_ACCOUNT: self.entry.data[CONF_ACCOUNT]}
elif user_input is not None:
try:
res = await self.hass.async_add_executor_job(validate_input, user_input)
if res[0] is not None:
name = str(res[0]["personaname"])
else:
errors = {"base": "invalid_account"}
except (steam.api.HTTPError, steam.api.HTTPTimeoutError) as ex:
errors = {"base": "cannot_connect"}
if "403" in str(ex):
errors = {"base": "invalid_auth"}
except Exception as ex: # pylint:disable=broad-except
LOGGER.exception("Unknown exception: %s", ex)
errors = {"base": "unknown"}
if not errors:
entry = await self.async_set_unique_id(user_input[CONF_ACCOUNT])
if entry and self.source == config_entries.SOURCE_REAUTH:
self.hass.config_entries.async_update_entry(entry, data=user_input)
await self.hass.config_entries.async_reload(entry.entry_id)
return self.async_abort(reason="reauth_successful")
self._abort_if_unique_id_configured()
if self.source == config_entries.SOURCE_IMPORT:
accounts_data = {
CONF_ACCOUNTS: {
acc["steamid"]: {
"name": acc["personaname"],
"enabled": True,
}
for acc in res
}
}
user_input.pop(CONF_ACCOUNTS)
else:
accounts_data = {
CONF_ACCOUNTS: {
user_input[CONF_ACCOUNT]: {"name": name, "enabled": True}
}
}
return self.async_create_entry(
title=name or DEFAULT_NAME,
data=user_input,
options=accounts_data,
)
user_input = user_input or {}
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(
CONF_API_KEY, default=user_input.get(CONF_API_KEY) or ""
): str,
vol.Required(
CONF_ACCOUNT, default=user_input.get(CONF_ACCOUNT) or ""
): str,
}
),
errors=errors,
)
async def async_step_import(self, import_config: ConfigType) -> FlowResult:
"""Import a config entry from configuration.yaml."""
for entry in self._async_current_entries():
if entry.data[CONF_API_KEY] == import_config[CONF_API_KEY]:
return self.async_abort(reason="already_configured")
LOGGER.warning(
"Steam yaml config in now deprecated and has been imported. "
"Please remove it from your config"
)
import_config[CONF_ACCOUNT] = import_config[CONF_ACCOUNTS][0]
return await self.async_step_user(import_config)
async def async_step_reauth(self, user_input: dict[str, str]) -> FlowResult:
"""Handle a reauthorization flow request."""
self.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, str] | None = None
) -> FlowResult:
"""Confirm reauth dialog."""
if user_input is not None:
return await self.async_step_user()
self._set_confirm_only()
return self.async_show_form(step_id="reauth_confirm")
class SteamOptionsFlowHandler(config_entries.OptionsFlow):
"""Handle Steam client options."""
def __init__(self, entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.entry = entry
self.options = dict(entry.options)
async def async_step_init(
self, user_input: dict[str, dict[str, str]] | None = None
) -> FlowResult:
"""Manage Steam options."""
if user_input is not None:
await self.hass.config_entries.async_unload(self.entry.entry_id)
for k in self.options[CONF_ACCOUNTS]:
if (
self.options[CONF_ACCOUNTS][k]["enabled"]
and k not in user_input[CONF_ACCOUNTS]
and (
entity_id := er.async_get(self.hass).async_get_entity_id(
Platform.SENSOR, DOMAIN, f"sensor.steam_{k}"
)
)
):
er.async_get(self.hass).async_remove(entity_id)
channel_data = {
CONF_ACCOUNTS: {
k: {
"name": v["name"],
"enabled": k in user_input[CONF_ACCOUNTS],
}
for k, v in self.options[CONF_ACCOUNTS].items()
if k in user_input[CONF_ACCOUNTS]
}
}
await self.hass.config_entries.async_reload(self.entry.entry_id)
return self.async_create_entry(title="", data=channel_data)
try:
users = {
name["steamid"]: {"name": name["personaname"], "enabled": False}
for name in await self.hass.async_add_executor_job(self.get_accounts)
}
except steam.api.HTTPTimeoutError:
users = self.options[CONF_ACCOUNTS]
_users = users | self.options[CONF_ACCOUNTS]
self.options[CONF_ACCOUNTS] = {
k: v
for k, v in _users.items()
if k in users or self.options[CONF_ACCOUNTS][k]["enabled"]
}
options = {
vol.Required(
CONF_ACCOUNTS,
default={
k
for k in self.options[CONF_ACCOUNTS]
if self.options[CONF_ACCOUNTS][k]["enabled"]
},
): cv.multi_select(
{k: v["name"] for k, v in self.options[CONF_ACCOUNTS].items()}
),
}
return self.async_show_form(step_id="init", data_schema=vol.Schema(options))
def get_accounts(self) -> list[dict[str, str | int]]:
"""Get accounts."""
interface = steam.api.interface("ISteamUser")
friends = interface.GetFriendList(steamid=self.entry.data[CONF_ACCOUNT])
friends = friends["friendslist"]["friends"]
_users_str = [user["steamid"] for user in friends]
names = interface.GetPlayerSummaries(steamids=_users_str)
return names["response"]["players"]["player"]

35
homeassistant/components/steam_online/const.py Normal file
View File

@ -0,0 +1,35 @@
"""Steam constants."""
import logging
from typing import Final
CONF_ACCOUNT = "account"
CONF_ACCOUNTS = "accounts"
DATA_KEY_COORDINATOR = "coordinator"
DEFAULT_NAME = "Steam"
DOMAIN: Final = "steam_online"
LOGGER = logging.getLogger(__package__)
STATE_OFFLINE = "offline"
STATE_ONLINE = "online"
STATE_BUSY = "busy"
STATE_AWAY = "away"
STATE_SNOOZE = "snooze"
STATE_LOOKING_TO_TRADE = "looking_to_trade"
STATE_LOOKING_TO_PLAY = "looking_to_play"
STEAM_STATUSES = {
0: STATE_OFFLINE,
1: STATE_ONLINE,
2: STATE_BUSY,
3: STATE_AWAY,
4: STATE_SNOOZE,
5: STATE_LOOKING_TO_TRADE,
6: STATE_LOOKING_TO_PLAY,
}
STEAM_API_URL = "https://steamcdn-a.akamaihd.net/steam/apps/"
STEAM_HEADER_IMAGE_FILE = "header.jpg"
STEAM_MAIN_IMAGE_FILE = "capsule_616x353.jpg"
STEAM_ICON_URL = (
"https://steamcdn-a.akamaihd.net/steamcommunity/public/images/apps/%d/%s.jpg"
)

71
homeassistant/components/steam_online/coordinator.py Normal file
View File

@ -0,0 +1,71 @@
"""Data update coordinator for the Steam integration."""
from __future__ import annotations
from datetime import timedelta
import steam
from steam.api import _interface_method as INTMethod
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_ACCOUNTS, DOMAIN, LOGGER
class SteamDataUpdateCoordinator(DataUpdateCoordinator):
"""Data update coordinator for the Steam integration."""
config_entry: ConfigEntry
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the coordinator."""
super().__init__(
hass=hass,
logger=LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=30),
)
self.game_icons: dict = {}
self.player_interface: INTMethod = None
self.user_interface: INTMethod = None
steam.api.key.set(self.config_entry.data[CONF_API_KEY])
def _update(self) -> dict[str, dict[str, str | int]]:
"""Fetch data from API endpoint."""
accounts = self.config_entry.options[CONF_ACCOUNTS]
_ids = [k for k in accounts if accounts[k]["enabled"]]
if not self.user_interface or not self.player_interface:
self.user_interface = steam.api.interface("ISteamUser")
self.player_interface = steam.api.interface("IPlayerService")
if not self.game_icons:
for _id in _ids:
res = self.player_interface.GetOwnedGames(
steamid=_id, include_appinfo=1
)["response"]
self.game_icons = self.game_icons | {
game["appid"]: game["img_icon_url"] for game in res.get("games", {})
}
response = self.user_interface.GetPlayerSummaries(steamids=_ids)
players = {
player["steamid"]: player
for player in response["response"]["players"]["player"]
if player["steamid"] in _ids
}
for k in players:
data = self.player_interface.GetSteamLevel(steamid=players[k]["steamid"])
data = data["response"]
players[k]["level"] = data["player_level"]
return players
async def _async_update_data(self) -> dict[str, dict[str, str | int]]:
"""Send request to the executor."""
try:
return await self.hass.async_add_executor_job(self._update)
except (steam.api.HTTPError, steam.api.HTTPTimeoutError) as ex:
if "401" in str(ex):
raise ConfigEntryAuthFailed from ex
raise UpdateFailed(ex) from ex

3
homeassistant/components/steam_online/manifest.json
View File

@ -1,9 +1,10 @@
{
"domain": "steam_online",
"name": "Steam",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/steam_online",
"requirements": ["steamodd==4.21"],
"codeowners": [],
"codeowners": ["@tkdrob"],
"iot_class": "cloud_polling",
"loggers": ["steam"]
}

277
homeassistant/components/steam_online/sensor.py
View File

@ -1,43 +1,37 @@
"""Sensor for Steam account status."""
from __future__ import annotations
from datetime import timedelta
import logging
from time import mktime
from datetime import datetime
from time import localtime, mktime
import steam
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
from homeassistant.components.sensor import (
PLATFORM_SCHEMA,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import track_time_interval
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateType
from homeassistant.util.dt import utc_from_timestamp
_LOGGER = logging.getLogger(__name__)
CONF_ACCOUNTS = "accounts"
ICON = "mdi:steam"
STATE_OFFLINE = "offline"
STATE_ONLINE = "online"
STATE_BUSY = "busy"
STATE_AWAY = "away"
STATE_SNOOZE = "snooze"
STATE_LOOKING_TO_TRADE = "looking_to_trade"
STATE_LOOKING_TO_PLAY = "looking_to_play"
STEAM_API_URL = "https://steamcdn-a.akamaihd.net/steam/apps/"
STEAM_HEADER_IMAGE_FILE = "header.jpg"
STEAM_MAIN_IMAGE_FILE = "capsule_616x353.jpg"
STEAM_ICON_URL = (
"https://steamcdn-a.akamaihd.net/steamcommunity/public/images/apps/%d/%s.jpg"
from . import SteamEntity
from .const import (
CONF_ACCOUNTS,
DOMAIN,
STEAM_API_URL,
STEAM_HEADER_IMAGE_FILE,
STEAM_ICON_URL,
STEAM_MAIN_IMAGE_FILE,
STEAM_STATUSES,
)
from .coordinator import SteamDataUpdateCoordinator
# Deprecated in Home Assistant 2022.5
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_API_KEY): cv.string,
@ -45,186 +39,89 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
}
)
APP_LIST_KEY = "steam_online.app_list"
BASE_INTERVAL = timedelta(minutes=1)
PARALLEL_UPDATES = 1
def setup_platform(
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Twitch sensor from 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:
"""Set up the Steam platform."""
steam.api.key.set(config[CONF_API_KEY])
# Initialize steammods app list before creating sensors
# to benefit from internal caching of the list.
hass.data[APP_LIST_KEY] = steam.apps.app_list()
entities = [SteamSensor(account, steam) for account in config[CONF_ACCOUNTS]]
if not entities:
return
add_entities(entities, True)
# Only one sensor update once every 60 seconds to avoid
# flooding steam and getting disconnected.
entity_next = 0
@callback
def do_update(time):
nonlocal entity_next
entities[entity_next].async_schedule_update_ha_state(True)
entity_next = (entity_next + 1) % len(entities)
track_time_interval(hass, do_update, BASE_INTERVAL)
async_add_entities(
SteamSensor(hass.data[DOMAIN][entry.entry_id], account)
for account in entry.options[CONF_ACCOUNTS]
if entry.options[CONF_ACCOUNTS][account]["enabled"]
)
class SteamSensor(SensorEntity):
class SteamSensor(SteamEntity, SensorEntity):
"""A class for the Steam account."""
def __init__(self, account, steamod):
def __init__(self, coordinator: SteamDataUpdateCoordinator, account: str) -> None:
"""Initialize the sensor."""
self._steamod = steamod
self._account = account
self._profile = None
self._game = None
self._game_id = None
self._extra_game_info = None
self._state = None
self._name = None
self._avatar = None
self._last_online = None
self._level = None
self._owned_games = None
super().__init__(coordinator)
self.entity_description = SensorEntityDescription(
key=account,
name=f"steam_{account}",
icon="mdi:steam",
)
self._attr_unique_id = f"sensor.steam_{account}"
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def entity_id(self):
"""Return the entity ID."""
return f"sensor.steam_{self._account}"
@property
def native_value(self):
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self._state
@property
def should_poll(self):
"""Turn off polling, will do ourselves."""
return False
def update(self):
"""Update device state."""
try:
self._profile = self._steamod.user.profile(self._account)
# Only if need be, get the owned games
if not self._owned_games:
self._owned_games = self._steamod.api.interface(
"IPlayerService"
).GetOwnedGames(steamid=self._account, include_appinfo=1)
self._game = self._get_current_game()
self._game_id = self._profile.current_game[0]
self._extra_game_info = self._get_game_info()
self._state = {
1: STATE_ONLINE,
2: STATE_BUSY,
3: STATE_AWAY,
4: STATE_SNOOZE,
5: STATE_LOOKING_TO_TRADE,
6: STATE_LOOKING_TO_PLAY,
}.get(self._profile.status, STATE_OFFLINE)
self._name = self._profile.persona
self._avatar = self._profile.avatar_medium
self._last_online = self._get_last_online()
self._level = self._profile.level
except self._steamod.api.HTTPTimeoutError as error:
_LOGGER.warning(error)
self._game = None
self._game_id = None
self._state = None
self._name = None
self._avatar = None
self._last_online = None
self._level = None
def _get_current_game(self):
"""Gather current game name from APP ID."""
if game_extra_info := self._profile.current_game[2]:
return game_extra_info
if not (game_id := self._profile.current_game[0]):
return None
app_list = self.hass.data[APP_LIST_KEY]
try:
_, res = app_list[game_id]
return res
except KeyError:
pass
# Try reloading the app list, must be a new app
app_list = self._steamod.apps.app_list()
self.hass.data[APP_LIST_KEY] = app_list
try:
_, res = app_list[game_id]
return res
except KeyError:
pass
_LOGGER.error("Unable to find name of app with ID=%s", game_id)
return repr(game_id)
def _get_game_info(self):
if (game_id := self._profile.current_game[0]) is not None:
for game in self._owned_games["response"]["games"]:
if game["appid"] == game_id:
return game
return None
def _get_last_online(self):
"""Convert last_online from the steam module into timestamp UTC."""
last_online = utc_from_timestamp(mktime(self._profile.last_online))
if last_online:
return last_online
if self.entity_description.key in self.coordinator.data:
player = self.coordinator.data[self.entity_description.key]
return STEAM_STATUSES[player["personastate"]]
return None
@property
def extra_state_attributes(self):
"""Return the state attributes."""
attr = {}
if self._game is not None:
attr["game"] = self._game
if self._game_id is not None:
attr["game_id"] = self._game_id
game_url = f"{STEAM_API_URL}{self._game_id}/"
attr["game_image_header"] = f"{game_url}{STEAM_HEADER_IMAGE_FILE}"
attr["game_image_main"] = f"{game_url}{STEAM_MAIN_IMAGE_FILE}"
if self._extra_game_info is not None and self._game_id is not None:
attr["game_icon"] = STEAM_ICON_URL % (
self._game_id,
self._extra_game_info["img_icon_url"],
)
if self._last_online is not None:
attr["last_online"] = self._last_online
if self._level is not None:
attr["level"] = self._level
return attr
def extra_state_attributes(self) -> dict[str, str | datetime]:
"""Return the state attributes of the sensor."""
if self.entity_description.key not in self.coordinator.data:
return {}
player = self.coordinator.data[self.entity_description.key]
@property
def entity_picture(self):
"""Avatar of the account."""
return self._avatar
attrs: dict[str, str | datetime] = {}
if game := player.get("gameextrainfo"):
attrs["game"] = game
if game_id := player.get("gameid"):
attrs["game_id"] = game_id
game_url = f"{STEAM_API_URL}{player['gameid']}/"
attrs["game_image_header"] = f"{game_url}{STEAM_HEADER_IMAGE_FILE}"
attrs["game_image_main"] = f"{game_url}{STEAM_MAIN_IMAGE_FILE}"
if info := self._get_game_icon(player):
attrs["game_icon"] = STEAM_ICON_URL % (
game_id,
info,
)
self._attr_name = player["personaname"]
self._attr_entity_picture = player["avatarmedium"]
if last_online := player.get("lastlogoff"):
attrs["last_online"] = utc_from_timestamp(mktime(localtime(last_online)))
if level := self.coordinator.data[self.entity_description.key]["level"]:
attrs["level"] = level
return attrs
@property
def icon(self):
"""Return the icon to use in the frontend."""
return ICON
def _get_game_icon(self, player: dict) -> str | None:
"""Get game icon identifier."""
if player.get("gameid") in self.coordinator.game_icons:
return self.coordinator.game_icons[player["gameid"]]
# Reset game icons to have coordinator get id for new game
self.coordinator.game_icons = {}
return None

36
homeassistant/components/steam_online/strings.json Normal file
View File

@ -0,0 +1,36 @@
{
"config": {
"step": {
"user": {
"description": "Use https://steamid.io to find your Steam account ID",
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]",
"account": "Steam account ID"
}
},
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "The Steam integration needs to be manually re-authenticated\n\nYou can find your key here: https://steamcommunity.com/dev/apikey"
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_account": "Invalid account ID",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"options": {
"step": {
"init": {
"data": {
"accounts": "Names of accounts to be monitored"
}
}
}
}
}

36
homeassistant/components/steam_online/translations/en.json Normal file
View File

@ -0,0 +1,36 @@
{
"config": {
"abort": {
"already_configured": "Service is already configured",
"reauth_successful": "Re-authentication was successful"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"invalid_account": "Invalid account ID",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"api_key": "API Key",
"account": "Steam account ID"
},
"description": "Documentation: https://www.home-assistant.io/integrations/steam_online\n\nUse https://steamid.io/ to find your Steam account ID"
},
"reauth_confirm": {
"title": "Reauthenticate Integration",
"description": "The Steam integration needs to be manually re-authenticated\n\nYou can find your key here: https://steamcommunity.com/dev/apikey"
}
}
},
"options": {
"step": {
"init": {
"data": {
"accounts": "Names of accounts to be monitored"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -327,6 +327,7 @@ FLOWS = {
"squeezebox",
"srp_energy",
"starline",
"steam_online",
"steamist",
"stookalert",
"subaru",

3
requirements_test_all.txt
View File

@ -1446,6 +1446,9 @@ starline==0.1.5
# homeassistant.components.statsd
statsd==3.2.1
# homeassistant.components.steam_online
steamodd==4.21
# homeassistant.components.stookalert
stookalert==0.1.4

125
tests/components/steam_online/__init__.py Normal file
View File

@ -0,0 +1,125 @@
"""Tests for Steam integration."""
from unittest.mock import patch
from homeassistant.components.steam_online import DOMAIN
from homeassistant.components.steam_online.const import CONF_ACCOUNT, CONF_ACCOUNTS
from homeassistant.const import CONF_API_KEY, CONF_NAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
API_KEY = "abc123"
ACCOUNT_1 = "1234567890"
ACCOUNT_2 = "1234567891"
ACCOUNT_NAME_1 = "testaccount1"
ACCOUNT_NAME_2 = "testaccount2"
CONF_DATA = {
CONF_API_KEY: API_KEY,
CONF_ACCOUNT: ACCOUNT_1,
}
CONF_OPTIONS = {
CONF_ACCOUNTS: {
ACCOUNT_1: {
CONF_NAME: ACCOUNT_NAME_1,
"enabled": True,
}
}
}
CONF_OPTIONS_2 = {
CONF_ACCOUNTS: {
ACCOUNT_1: {
CONF_NAME: ACCOUNT_NAME_1,
"enabled": True,
},
ACCOUNT_2: {
CONF_NAME: ACCOUNT_NAME_2,
"enabled": True,
},
}
}
CONF_IMPORT_OPTIONS = {
CONF_ACCOUNTS: {
ACCOUNT_1: {
CONF_NAME: ACCOUNT_NAME_1,
"enabled": True,
},
ACCOUNT_2: {
CONF_NAME: ACCOUNT_NAME_2,
"enabled": True,
},
}
}
CONF_IMPORT_DATA = {CONF_API_KEY: API_KEY, CONF_ACCOUNTS: [ACCOUNT_1, ACCOUNT_2]}
def create_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Add config entry in Home Assistant."""
entry = MockConfigEntry(
domain=DOMAIN,
data=CONF_DATA,
options=CONF_OPTIONS,
unique_id=ACCOUNT_1,
)
entry.add_to_hass(hass)
return entry
class MockedUserInterfaceNull:
"""Mocked user interface returning no players."""
def GetPlayerSummaries(self, steamids: str) -> dict:
"""Get player summaries."""
return {"response": {"players": {"player": [None]}}}
class MockedInterface(dict):
"""Mocked interface."""
def IPlayerService(self) -> None:
"""Mock iplayerservice."""
def ISteamUser(self) -> None:
"""Mock iSteamUser."""
def GetFriendList(self, steamid: str) -> dict:
"""Get friend list."""
return {"friendslist": {"friends": [{"steamid": ACCOUNT_2}]}}
def GetPlayerSummaries(self, steamids: str) -> dict:
"""Get player summaries."""
return {
"response": {
"players": {
"player": [
{"steamid": ACCOUNT_1, "personaname": ACCOUNT_NAME_1},
{"steamid": ACCOUNT_2, "personaname": ACCOUNT_NAME_2},
]
}
}
}
def GetOwnedGames(self, steamid: str, include_appinfo: int) -> dict:
"""Get owned games."""
return {
"response": {"game_count": 1},
"games": [{"appid": 1, "img_icon_url": "1234567890"}],
}
def GetSteamLevel(self, steamid: str) -> dict:
"""Get steam level."""
return {"response": {"player_level": 10}}
def patch_interface() -> MockedInterface:
"""Patch interface."""
return patch("steam.api.interface", return_value=MockedInterface())
def patch_user_interface_null() -> MockedUserInterfaceNull:
"""Patch player interface with no players."""
return patch("steam.api.interface", return_value=MockedUserInterfaceNull())

222
tests/components/steam_online/test_config_flow.py Normal file
View File

@ -0,0 +1,222 @@
"""Test Steam config flow."""
import steam
from homeassistant import data_entry_flow
from homeassistant.components.steam_online.const import CONF_ACCOUNTS, DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_REAUTH, SOURCE_USER
from homeassistant.const import CONF_API_KEY, CONF_SOURCE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import (
ACCOUNT_1,
ACCOUNT_2,
ACCOUNT_NAME_1,
CONF_DATA,
CONF_IMPORT_DATA,
CONF_IMPORT_OPTIONS,
CONF_OPTIONS,
CONF_OPTIONS_2,
create_entry,
patch_interface,
patch_user_interface_null,
)
async def test_flow_user(hass: HomeAssistant) -> None:
"""Test user initialized flow."""
with patch_interface():
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_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == ACCOUNT_NAME_1
assert result["data"] == CONF_DATA
assert result["options"] == CONF_OPTIONS
assert result["result"].unique_id == ACCOUNT_1
async def test_flow_user_cannot_connect(hass: HomeAssistant) -> None:
"""Test user initialized flow with unreachable server."""
with patch_interface() as servicemock:
servicemock.side_effect = steam.api.HTTPTimeoutError
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "cannot_connect"
async def test_flow_user_invalid_auth(hass: HomeAssistant) -> None:
"""Test user initialized flow with invalid authentication."""
with patch_interface() as servicemock:
servicemock.side_effect = steam.api.HTTPError("403")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "invalid_auth"
async def test_flow_user_invalid_account(hass: HomeAssistant) -> None:
"""Test user initialized flow with invalid account ID."""
with patch_user_interface_null():
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "invalid_account"
async def test_flow_user_unknown(hass: HomeAssistant) -> None:
"""Test user initialized flow with unknown error."""
with patch_interface() as servicemock:
servicemock.side_effect = Exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "unknown"
async def test_flow_user_already_configured(hass: HomeAssistant) -> None:
"""Test user initialized flow with duplicate account."""
create_entry(hass)
with patch_interface():
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_flow_reauth(hass: HomeAssistant) -> None:
"""Test reauth step."""
entry = create_entry(hass)
with patch_interface():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
CONF_SOURCE: SOURCE_REAUTH,
"entry_id": entry.entry_id,
"unique_id": entry.unique_id,
},
data=CONF_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
new_conf = CONF_DATA | {CONF_API_KEY: "1234567890"}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=new_conf,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful"
assert entry.data == new_conf
async def test_flow_import(hass: HomeAssistant) -> None:
"""Test import step."""
with patch_interface():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=CONF_IMPORT_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == ACCOUNT_NAME_1
assert result["data"] == CONF_DATA
assert result["options"] == CONF_IMPORT_OPTIONS
assert result["result"].unique_id == ACCOUNT_1
async def test_flow_import_already_configured(hass: HomeAssistant) -> None:
"""Test import step already configured."""
create_entry(hass)
with patch_interface():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=CONF_IMPORT_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_options_flow(hass: HomeAssistant) -> None:
"""Test updating options."""
entry = create_entry(hass)
with patch_interface():
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.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_ACCOUNTS: [ACCOUNT_1, ACCOUNT_2]},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == CONF_OPTIONS_2
assert len(er.async_get(hass).entities) == 2
async def test_options_flow_deselect(hass: HomeAssistant) -> None:
"""Test deselecting user."""
entry = create_entry(hass)
with patch_interface():
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.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_ACCOUNTS: []},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == {CONF_ACCOUNTS: {}}
assert len(er.async_get(hass).entities) == 0
async def test_options_flow_timeout(hass: HomeAssistant) -> None:
"""Test updating options timeout getting friends list."""
entry = create_entry(hass)
with patch_interface() as servicemock:
servicemock.side_effect = steam.api.HTTPTimeoutError
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_ACCOUNTS: [ACCOUNT_1]},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == CONF_OPTIONS

52
tests/components/steam_online/test_init.py Normal file
View File

@ -0,0 +1,52 @@
"""Tests for the Steam component."""
import steam
from homeassistant.components.steam_online.const import DEFAULT_NAME, DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import create_entry, patch_interface
async def test_setup(hass: HomeAssistant) -> None:
"""Test unload."""
entry = create_entry(hass)
with patch_interface():
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert entry.state == ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
assert not hass.data.get(DOMAIN)
async def test_async_setup_entry_auth_failed(hass: HomeAssistant) -> None:
"""Test that it throws ConfigEntryAuthFailed when authentication fails."""
entry = create_entry(hass)
with patch_interface() as interface:
interface.side_effect = steam.api.HTTPError("401")
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_ERROR
assert not hass.data.get(DOMAIN)
async def test_device_info(hass: HomeAssistant) -> None:
"""Test device info."""
entry = create_entry(hass)
with patch_interface():
await hass.config_entries.async_setup(entry.entry_id)
device_registry = await dr.async_get_registry(hass)
await hass.async_block_till_done()
device = device_registry.async_get_device({(DOMAIN, entry.entry_id)})
assert device.configuration_url == "https://store.steampowered.com"
assert device.entry_type == dr.DeviceEntryType.SERVICE
assert device.identifiers == {(DOMAIN, entry.entry_id)}
assert device.manufacturer == DEFAULT_NAME
assert device.name == DEFAULT_NAME