diff --git a/homeassistant/components/steam_online/config_flow.py b/homeassistant/components/steam_online/config_flow.py index 6da975bcd68b..fc5313376e80 100644 --- a/homeassistant/components/steam_online/config_flow.py +++ b/homeassistant/components/steam_online/config_flow.py @@ -1,6 +1,7 @@ """Config flow for Steam integration.""" from collections.abc import Iterator, Mapping +import logging from typing import Any, override import steam.api @@ -17,9 +18,12 @@ from homeassistant.const import CONF_API_KEY, CONF_NAME, Platform from homeassistant.core import callback from homeassistant.helpers import config_validation as cv, entity_registry as er -from .const import CONF_ACCOUNT, CONF_ACCOUNTS, DOMAIN, LOGGER, PLACEHOLDERS +from .const import CONF_ACCOUNT, CONF_ACCOUNTS, DOMAIN, PLACEHOLDERS from .coordinator import SteamConfigEntry +_LOGGER = logging.getLogger(__name__) + + # To avoid too long request URIs, the amount of ids to request is limited MAX_IDS_TO_REQUEST = 275 @@ -75,8 +79,8 @@ class SteamFlowHandler(ConfigFlow, domain=DOMAIN): errors["base"] = ( "invalid_auth" if "403" in str(ex) else "cannot_connect" ) - except Exception: # noqa: BLE001 - LOGGER.exception("Unknown exception") + except Exception: + _LOGGER.exception("Unknown exception") errors["base"] = "unknown" if not errors: return self.async_create_entry( @@ -129,8 +133,8 @@ class SteamFlowHandler(ConfigFlow, domain=DOMAIN): errors["base"] = ( "invalid_auth" if "403" in str(ex) else "cannot_connect" ) - except Exception: # noqa: BLE001 - LOGGER.exception("Unknown exception") + except Exception: + _LOGGER.exception("Unknown exception") errors["base"] = "unknown" if not errors: @@ -166,7 +170,6 @@ class SteamOptionsFlowHandler(OptionsFlowWithReload): ) -> ConfigFlowResult: """Manage Steam options.""" if user_input is not None: - await self.hass.config_entries.async_unload(self.config_entry.entry_id) for _id in self.options[CONF_ACCOUNTS]: if _id not in user_input[CONF_ACCOUNTS] and ( entity_id := er.async_get(self.hass).async_get_entity_id( diff --git a/homeassistant/components/steam_online/const.py b/homeassistant/components/steam_online/const.py index ec5fa3d475f8..14654da3b765 100644 --- a/homeassistant/components/steam_online/const.py +++ b/homeassistant/components/steam_online/const.py @@ -1,6 +1,5 @@ """Steam constants.""" -import logging from typing import Final CONF_ACCOUNT = "account" @@ -10,7 +9,6 @@ DATA_KEY_COORDINATOR = "coordinator" DEFAULT_NAME = "Steam" DOMAIN: Final = "steam_online" -LOGGER = logging.getLogger(__package__) PLACEHOLDERS = { "api_key_url": "https://steamcommunity.com/dev/apikey", diff --git a/homeassistant/components/steam_online/coordinator.py b/homeassistant/components/steam_online/coordinator.py index 72cf04b85648..7c3df2b82c4a 100644 --- a/homeassistant/components/steam_online/coordinator.py +++ b/homeassistant/components/steam_online/coordinator.py @@ -1,10 +1,11 @@ """Data update coordinator for the Steam integration.""" +from dataclasses import dataclass from datetime import timedelta +import logging from typing import override import steam.api -from steam.api import _interface_method as INTMethod from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_API_KEY @@ -12,65 +13,116 @@ 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 +from .const import CONF_ACCOUNTS, DOMAIN type SteamConfigEntry = ConfigEntry[SteamDataUpdateCoordinator] +_LOGGER = logging.getLogger(__name__) -class SteamDataUpdateCoordinator( - DataUpdateCoordinator[dict[str, dict[str, str | int]]] -): + +@dataclass(kw_only=True, frozen=True) +class PlayerData: + """Steam player data.""" + + steamid: str + communityvisibilitystate: int + profilestate: int + personaname: str + commentpermission: int | None = None + profileurl: str + avatar: str + avatarmedium: str + avatarfull: str + avatarhash: str + lastlogoff: int + personastate: int + realname: str | None = None + primaryclanid: str | None = None + timecreated: int | None = None + personastateflags: int + loccountrycode: str | None = None + locstatecode: str | None = None + loccityid: int | None = None + gameextrainfo: str | None = None + gameid: str | None = None + level: int | None = None + + +class SteamDataUpdateCoordinator(DataUpdateCoordinator[dict[str, PlayerData]]): """Data update coordinator for the Steam integration.""" config_entry: SteamConfigEntry + user_interface: steam.api.interface + player_interface: steam.api.interface def __init__(self, hass: HomeAssistant, config_entry: SteamConfigEntry) -> None: """Initialize the coordinator.""" super().__init__( hass=hass, - logger=LOGGER, + logger=_LOGGER, config_entry=config_entry, name=DOMAIN, update_interval=timedelta(seconds=30), ) - self.game_icons: dict[int, str] = {} - self.player_interface: INTMethod = None - self.user_interface: INTMethod = None - steam.api.key.set(self.config_entry.data[CONF_API_KEY]) + self.game_icons: dict[str, str] = {} - def _update(self) -> dict[str, dict[str, str | int]]: + @override + async def _async_setup(self) -> None: + """Set up the coordinator.""" + + steam.api.key.set(self.config_entry.data[CONF_API_KEY]) + self.user_interface = steam.api.interface("ISteamUser") + self.player_interface = steam.api.interface("IPlayerService") + + def _update(self) -> dict[str, PlayerData]: """Fetch data from API endpoint.""" accounts = self.config_entry.options[CONF_ACCOUNTS] _ids = list(accounts) - 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 + player["steamid"]: PlayerData( + **player, + level=self.player_interface.GetSteamLevel(steamid=player["steamid"])[ + "response" + ].get("player_level"), + ) for player in response["response"]["players"]["player"] if player["steamid"] in _ids } - for value in players.values(): - data = self.player_interface.GetSteamLevel(steamid=value["steamid"]) - value["level"] = data["response"].get("player_level") + + for player in players.values(): + if player.gameid and player.gameid not in self.game_icons: + games = self.player_interface.GetOwnedGames( + steamid=player.steamid, + include_appinfo=1, + include_played_free_games=True, + )["response"].get("games", []) + self.game_icons.update( + {str(game["appid"]): game["img_icon_url"] for game in games} + ) + return players @override - async def _async_update_data(self) -> dict[str, dict[str, str | int]]: + async def _async_update_data(self) -> dict[str, PlayerData]: """Send request to the executor.""" try: return await self.hass.async_add_executor_job(self._update) + except steam.api.HTTPTimeoutError as ex: + raise UpdateFailed( + translation_domain=DOMAIN, + translation_key="timeout_exception", + ) from ex except steam.api.HTTPError as ex: - if "401" in str(ex): - raise ConfigEntryAuthFailed from ex - raise UpdateFailed(ex) from ex + _LOGGER.debug("Full exception:", exc_info=True) + if "401" in str(ex) or "403" in str(ex): + raise ConfigEntryAuthFailed( + translation_domain=DOMAIN, + translation_key="auth_exception", + ) from ex + raise UpdateFailed( + translation_domain=DOMAIN, + translation_key="request_exception", + ) from ex diff --git a/homeassistant/components/steam_online/sensor.py b/homeassistant/components/steam_online/sensor.py index 636a8f034648..7749c3a5b29b 100644 --- a/homeassistant/components/steam_online/sensor.py +++ b/homeassistant/components/steam_online/sensor.py @@ -4,7 +4,7 @@ from collections.abc import Callable from dataclasses import dataclass from datetime import datetime from enum import StrEnum -from typing import Any, cast, override +from typing import Any, override from homeassistant.components.sensor import SensorEntity, SensorEntityDescription from homeassistant.core import HomeAssistant @@ -20,7 +20,7 @@ from .const import ( STEAM_MAIN_IMAGE_FILE, STEAM_STATUSES, ) -from .coordinator import SteamConfigEntry, SteamDataUpdateCoordinator +from .coordinator import PlayerData, SteamConfigEntry, SteamDataUpdateCoordinator from .entity import SteamEntity PARALLEL_UPDATES = 1 @@ -36,18 +36,18 @@ class SteamSensor(StrEnum): class SteamSensorEntityDescription(SensorEntityDescription): """Steam sensor description.""" - value_fn: Callable[[dict[str, Any]], StateType] - name_fn: Callable[[dict[str, Any]], str] - entity_picture_fn: Callable[[dict[str, Any]], str] | None = None + value_fn: Callable[[PlayerData], StateType] + name_fn: Callable[[PlayerData], str] + entity_picture_fn: Callable[[PlayerData], str] | None = None SENSOR_DESCRIPTIONS: tuple[SteamSensorEntityDescription, ...] = ( SteamSensorEntityDescription( key=SteamSensor.ACCOUNT, translation_key=SteamSensor.ACCOUNT, - value_fn=lambda x: STEAM_STATUSES[x["personastate"]], - name_fn=lambda x: x["personaname"], - entity_picture_fn=lambda x: x["avatarfull"], + value_fn=lambda x: STEAM_STATUSES[x.personastate], + name_fn=lambda x: x.personaname, + entity_picture_fn=lambda x: x.avatarfull, ), ) @@ -106,29 +106,27 @@ class SteamSensorEntity(SteamEntity, SensorEntity): player = self.coordinator.data[self._steamid] attrs: dict[str, str | int | datetime] = {} - if game := player.get("gameextrainfo"): + if game := player.gameextrainfo: attrs["game"] = game - if game_id := player.get("gameid"): + if game_id := player.gameid: attrs["game_id"] = game_id - game_url = f"{STEAM_API_URL}{player['gameid']}/" + 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"] = f"{STEAM_ICON_URL}{game_id}/{info}.jpg" - if last_online := cast(int | None, player.get("lastlogoff")): + if last_online := player.lastlogoff: attrs["last_online"] = dt_util.as_local( dt_util.utc_from_timestamp(last_online) ) - if level := self.coordinator.data[self._steamid]["level"]: + if level := self.coordinator.data[self._steamid].level: attrs["level"] = level return attrs - def _get_game_icon(self, player: dict) -> str | None: + def _get_game_icon(self, player: PlayerData) -> 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 = {} + if player.gameid is not None and player.gameid in self.coordinator.game_icons: + return self.coordinator.game_icons[player.gameid] return None @property diff --git a/homeassistant/components/steam_online/strings.json b/homeassistant/components/steam_online/strings.json index 7d244d6b9b9f..4d51800a505e 100644 --- a/homeassistant/components/steam_online/strings.json +++ b/homeassistant/components/steam_online/strings.json @@ -70,6 +70,17 @@ } } }, + "exceptions": { + "auth_exception": { + "message": "Failed to connect to Steam due to an authentication error" + }, + "request_exception": { + "message": "Failed to connect to Steam due to a request error" + }, + "timeout_exception": { + "message": "Failed to connect to Steam due to a request timeout" + } + }, "options": { "error": { "unauthorized": "Friends list restricted: Please refer to the documentation on how to see all other friends" diff --git a/tests/components/steam_online/snapshots/test_sensor.ambr b/tests/components/steam_online/snapshots/test_sensor.ambr index 598b4efa875d..26e5aa3a1b0b 100644 --- a/tests/components/steam_online/snapshots/test_sensor.ambr +++ b/tests/components/steam_online/snapshots/test_sensor.ambr @@ -42,6 +42,7 @@ : 'https://avatars.steamstatic.com/fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb_full.jpg', : 'Steam testaccount1', 'game': 'The Witcher: Enhanced Edition', + 'game_icon': 'https://steamcdn-a.akamaihd.net/steamcommunity/public/images/apps/20900/746d1cd48fb2e57d579b05b6e9eccba95859e549.jpg', 'game_id': '20900', 'game_image_header': 'https://steamcdn-a.akamaihd.net/steam/apps/20900/header.jpg', 'game_image_main': 'https://steamcdn-a.akamaihd.net/steam/apps/20900/capsule_616x353.jpg', diff --git a/tests/components/steam_online/test_config_flow.py b/tests/components/steam_online/test_config_flow.py index 2778664c5148..9c22c9ab09ef 100644 --- a/tests/components/steam_online/test_config_flow.py +++ b/tests/components/steam_online/test_config_flow.py @@ -64,7 +64,7 @@ async def test_flow_user( async def test_flow_user_errors( hass: HomeAssistant, mock_setup_entry: AsyncMock, - side_effect: Exception | dict[str, Any], + side_effect: type[Exception] | dict[str, Any], error_msg: str, steam_api: MagicMock, ) -> None: @@ -157,7 +157,7 @@ async def test_flow_reauth_errors( hass: HomeAssistant, config_entry: MockConfigEntry, steam_api: MagicMock, - side_effect: Exception | dict[str, Any], + side_effect: type[Exception] | dict[str, Any], error_msg: str, ) -> None: """Test reauth step with errors.""" diff --git a/tests/components/steam_online/test_init.py b/tests/components/steam_online/test_init.py index 7f626f6a99cc..89ea46f6a00b 100644 --- a/tests/components/steam_online/test_init.py +++ b/tests/components/steam_online/test_init.py @@ -35,30 +35,46 @@ async def test_setup( assert not hass.data.get(DOMAIN) +@pytest.mark.parametrize( + "side_effect", + [ + steam.api.HTTPError, + steam.api.HTTPTimeoutError, + ], +) async def test_setup_errors( hass: HomeAssistant, config_entry: MockConfigEntry, steam_api: MagicMock, + side_effect: type[Exception], ) -> None: """Test setup errors.""" config_entry.add_to_hass(hass) - steam_api.return_value.GetPlayerSummaries.side_effect = steam.api.HTTPError + steam_api.return_value.GetPlayerSummaries.side_effect = side_effect await hass.config_entries.async_setup(config_entry.entry_id) await hass.async_block_till_done() assert config_entry.state is ConfigEntryState.SETUP_RETRY +@pytest.mark.parametrize( + "side_effect", + [ + steam.api.HTTPError("Server connection failed: Forbidden (403)"), + steam.api.HTTPError("Server connection failed: Unauthorized (401)"), + ], +) async def test_setup_auth_failed( hass: HomeAssistant, config_entry: MockConfigEntry, steam_api: MagicMock, + side_effect: type[Exception], ) -> None: """Test that it throws ConfigEntryAuthFailed when authentication fails.""" config_entry.add_to_hass(hass) - steam_api.return_value.GetPlayerSummaries.side_effect = steam.api.HTTPError("401") + steam_api.return_value.GetPlayerSummaries.side_effect = side_effect await hass.config_entries.async_setup(config_entry.entry_id) await hass.async_block_till_done()