Add friend tracking to PlayStation Network (#149546)

This commit is contained in:
Manu 2025-07-30 15:10:30 +02:00 committed by GitHub
parent dd0b23afb0
commit ba4e7e50e0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 920 additions and 33 deletions

View File

@ -8,6 +8,7 @@ from homeassistant.core import HomeAssistant
from .const import CONF_NPSSO
from .coordinator import (
PlaystationNetworkConfigEntry,
PlaystationNetworkFriendDataCoordinator,
PlaystationNetworkGroupsUpdateCoordinator,
PlaystationNetworkRuntimeData,
PlaystationNetworkTrophyTitlesCoordinator,
@ -39,14 +40,33 @@ async def async_setup_entry(
groups = PlaystationNetworkGroupsUpdateCoordinator(hass, psn, entry)
await groups.async_config_entry_first_refresh()
friends = {}
for subentry_id, subentry in entry.subentries.items():
friend_coordinator = PlaystationNetworkFriendDataCoordinator(
hass, psn, entry, subentry
)
await friend_coordinator.async_config_entry_first_refresh()
friends[subentry_id] = friend_coordinator
entry.runtime_data = PlaystationNetworkRuntimeData(
coordinator, trophy_titles, groups
coordinator, trophy_titles, groups, friends
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
return True
async def _async_update_listener(
hass: HomeAssistant, entry: PlaystationNetworkConfigEntry
) -> None:
"""Handle update."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(
hass: HomeAssistant, entry: PlaystationNetworkConfigEntry
) -> bool:

View File

@ -10,13 +10,28 @@ from psnawp_api.core.psnawp_exceptions import (
PSNAWPInvalidTokenError,
PSNAWPNotFoundError,
)
from psnawp_api.models import User
from psnawp_api.utils.misc import parse_npsso_token
import voluptuous as vol
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import (
SOURCE_REAUTH,
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.const import CONF_NAME
from homeassistant.core import callback
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
)
from .const import CONF_NPSSO, DOMAIN, NPSSO_LINK, PSN_LINK
from .const import CONF_ACCOUNT_ID, CONF_NPSSO, DOMAIN, NPSSO_LINK, PSN_LINK
from .coordinator import PlaystationNetworkConfigEntry
from .helpers import PlaystationNetwork
_LOGGER = logging.getLogger(__name__)
@ -27,6 +42,14 @@ STEP_USER_DATA_SCHEMA = vol.Schema({vol.Required(CONF_NPSSO): str})
class PlaystationNetworkConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Playstation Network."""
@classmethod
@callback
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this integration."""
return {"friend": FriendSubentryFlowHandler}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@ -54,6 +77,15 @@ class PlaystationNetworkConfigFlow(ConfigFlow, domain=DOMAIN):
else:
await self.async_set_unique_id(user.account_id)
self._abort_if_unique_id_configured()
config_entries = self.hass.config_entries.async_entries(DOMAIN)
for entry in config_entries:
if user.account_id in {
subentry.unique_id for subentry in entry.subentries.values()
}:
return self.async_abort(
reason="already_configured_as_subentry"
)
return self.async_create_entry(
title=user.online_id,
data={CONF_NPSSO: npsso},
@ -132,3 +164,61 @@ class PlaystationNetworkConfigFlow(ConfigFlow, domain=DOMAIN):
"psn_link": PSN_LINK,
},
)
class FriendSubentryFlowHandler(ConfigSubentryFlow):
"""Handle subentry flow for adding a friend."""
friends_list: dict[str, User]
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Subentry user flow."""
config_entry: PlaystationNetworkConfigEntry = self._get_entry()
if user_input is not None:
config_entries = self.hass.config_entries.async_entries(DOMAIN)
if user_input[CONF_ACCOUNT_ID] in {
entry.unique_id for entry in config_entries
}:
return self.async_abort(reason="already_configured_as_entry")
for entry in config_entries:
if user_input[CONF_ACCOUNT_ID] in {
subentry.unique_id for subentry in entry.subentries.values()
}:
return self.async_abort(reason="already_configured")
return self.async_create_entry(
title=self.friends_list[user_input[CONF_ACCOUNT_ID]].online_id,
data={},
unique_id=user_input[CONF_ACCOUNT_ID],
)
self.friends_list = await self.hass.async_add_executor_job(
lambda: {
friend.account_id: friend
for friend in config_entry.runtime_data.user_data.psn.user.friends_list()
}
)
options = [
SelectOptionDict(
value=friend.account_id,
label=friend.online_id,
)
for friend in self.friends_list.values()
]
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Required(CONF_ACCOUNT_ID): SelectSelector(
SelectSelectorConfig(options=options)
)
}
),
user_input,
),
)

View File

@ -6,6 +6,7 @@ from psnawp_api.models.trophies import PlatformType
DOMAIN = "playstation_network"
CONF_NPSSO: Final = "npsso"
CONF_ACCOUNT_ID: Final = "account_id"
SUPPORTED_PLATFORMS = {
PlatformType.PS_VITA,

View File

@ -6,21 +6,30 @@ from abc import abstractmethod
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Any
from psnawp_api.core.psnawp_exceptions import (
PSNAWPAuthenticationError,
PSNAWPClientError,
PSNAWPError,
PSNAWPForbiddenError,
PSNAWPNotFoundError,
PSNAWPServerError,
)
from psnawp_api.models import User
from psnawp_api.models.group.group_datatypes import GroupDetails
from psnawp_api.models.trophies import TrophyTitle
from homeassistant.config_entries import ConfigEntry
from homeassistant.config_entries import ConfigEntry, ConfigSubentry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryError,
ConfigEntryNotReady,
)
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
from .const import CONF_ACCOUNT_ID, DOMAIN
from .helpers import PlaystationNetwork, PlaystationNetworkData
_LOGGER = logging.getLogger(__name__)
@ -35,6 +44,7 @@ class PlaystationNetworkRuntimeData:
user_data: PlaystationNetworkUserDataCoordinator
trophy_titles: PlaystationNetworkTrophyTitlesCoordinator
groups: PlaystationNetworkGroupsUpdateCoordinator
friends: dict[str, PlaystationNetworkFriendDataCoordinator]
class PlayStationNetworkBaseCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
@ -140,3 +150,78 @@ class PlaystationNetworkGroupsUpdateCoordinator(
if not group_info.group_id.startswith("~")
}
)
class PlaystationNetworkFriendDataCoordinator(
PlayStationNetworkBaseCoordinator[PlaystationNetworkData]
):
"""Friend status data update coordinator for PSN."""
user: User
profile: dict[str, Any]
def __init__(
self,
hass: HomeAssistant,
psn: PlaystationNetwork,
config_entry: PlaystationNetworkConfigEntry,
subentry: ConfigSubentry,
) -> None:
"""Initialize the Coordinator."""
self._update_interval = timedelta(
seconds=max(9 * len(config_entry.subentries), 180)
)
super().__init__(hass, psn, config_entry)
self.subentry = subentry
def _setup(self) -> None:
"""Set up the coordinator."""
self.user = self.psn.psn.user(account_id=self.subentry.data[CONF_ACCOUNT_ID])
self.profile = self.user.profile()
async def _async_setup(self) -> None:
"""Set up the coordinator."""
try:
await self.hass.async_add_executor_job(self._setup)
except PSNAWPNotFoundError as error:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="user_not_found",
translation_placeholders={"user": self.subentry.title},
) from error
except PSNAWPAuthenticationError as error:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="not_ready",
) from error
except (PSNAWPServerError, PSNAWPClientError) as error:
_LOGGER.debug("Update failed", exc_info=True)
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="update_failed",
) from error
def _update_data(self) -> PlaystationNetworkData:
"""Update friend status data."""
try:
return PlaystationNetworkData(
username=self.user.online_id,
account_id=self.user.account_id,
presence=self.user.get_presence(),
profile=self.profile,
)
except PSNAWPForbiddenError as error:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="user_profile_private",
translation_placeholders={"user": self.subentry.title},
) from error
except PSNAWPError:
raise
async def update_data(self) -> PlaystationNetworkData:
"""Update friend status data."""
return await self.hass.async_add_executor_job(self._update_data)

View File

@ -2,12 +2,14 @@
from typing import TYPE_CHECKING
from homeassistant.config_entries import ConfigSubentry
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import PlayStationNetworkBaseCoordinator
from .helpers import PlaystationNetworkData
class PlaystationNetworkServiceEntity(
@ -21,18 +23,32 @@ class PlaystationNetworkServiceEntity(
self,
coordinator: PlayStationNetworkBaseCoordinator,
entity_description: EntityDescription,
subentry: ConfigSubentry | None = None,
) -> None:
"""Initialize PlayStation Network Service Entity."""
super().__init__(coordinator)
if TYPE_CHECKING:
assert coordinator.config_entry.unique_id
self.entity_description = entity_description
self._attr_unique_id = (
f"{coordinator.config_entry.unique_id}_{entity_description.key}"
self.subentry = subentry
unique_id = (
subentry.unique_id
if subentry is not None and subentry.unique_id
else coordinator.config_entry.unique_id
)
self._attr_unique_id = f"{unique_id}_{entity_description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, coordinator.config_entry.unique_id)},
name=coordinator.psn.user.online_id,
identifiers={(DOMAIN, unique_id)},
name=(
coordinator.data.username
if isinstance(coordinator.data, PlaystationNetworkData)
else coordinator.psn.user.online_id
),
entry_type=DeviceEntryType.SERVICE,
manufacturer="Sony Interactive Entertainment",
)
if subentry:
self._attr_device_info.update(
DeviceInfo(via_device=(DOMAIN, coordinator.config_entry.unique_id))
)

View File

@ -38,7 +38,6 @@ class PlaystationNetworkData:
presence: dict[str, Any] = field(default_factory=dict)
username: str = ""
account_id: str = ""
availability: str = "unavailable"
active_sessions: dict[PlatformType, SessionData] = field(default_factory=dict)
registered_platforms: set[PlatformType] = field(default_factory=set)
trophy_summary: TrophySummary | None = None
@ -61,6 +60,7 @@ class PlaystationNetwork:
self.legacy_profile: dict[str, Any] | None = None
self.trophy_titles: list[TrophyTitle] = []
self._title_icon_urls: dict[str, str] = {}
self.friends_list: dict[str, User] | None = None
def _setup(self) -> None:
"""Setup PSN."""
@ -97,6 +97,7 @@ class PlaystationNetwork:
# check legacy platforms if owned
if LEGACY_PLATFORMS & data.registered_platforms:
self.legacy_profile = self.client.get_profile_legacy()
return data
async def get_data(self) -> PlaystationNetworkData:
@ -105,7 +106,6 @@ class PlaystationNetwork:
data.username = self.user.online_id
data.account_id = self.user.account_id
data.shareable_profile_link = self.shareable_profile_link
data.availability = data.presence["basicPresence"]["availability"]
if "platform" in data.presence["basicPresence"]["primaryPlatformInfo"]:
primary_platform = PlatformType(
@ -193,3 +193,17 @@ class PlaystationNetwork:
def normalize_title(name: str) -> str:
"""Normalize trophy title."""
return name.removesuffix("Trophies").removesuffix("Trophy Set").strip()
def get_game_title_info(presence: dict[str, Any]) -> dict[str, Any]:
"""Retrieve title info from presence."""
return (
next((title for title in game_title_info), {})
if (
game_title_info := presence.get("basicPresence", {}).get(
"gameTitleInfoList"
)
)
else {}
)

View File

@ -42,6 +42,13 @@
"availabletocommunicate": "mdi:cellphone",
"offline": "mdi:account-off-outline"
}
},
"now_playing": {
"default": "mdi:controller",
"state": {
"unknown": "mdi:controller-off",
"unavailable": "mdi:controller-off"
}
}
},
"image": {

View File

@ -5,18 +5,23 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from enum import StrEnum
from typing import TYPE_CHECKING
from homeassistant.components.image import ImageEntity, ImageEntityDescription
from homeassistant.config_entries import ConfigSubentry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .coordinator import (
PlayStationNetworkBaseCoordinator,
PlaystationNetworkConfigEntry,
PlaystationNetworkData,
PlaystationNetworkFriendDataCoordinator,
PlaystationNetworkUserDataCoordinator,
)
from .entity import PlaystationNetworkServiceEntity
from .helpers import get_game_title_info
PARALLEL_UPDATES = 0
@ -26,6 +31,7 @@ class PlaystationNetworkImage(StrEnum):
AVATAR = "avatar"
SHARE_PROFILE = "share_profile"
NOW_PLAYING_IMAGE = "now_playing_image"
@dataclass(kw_only=True, frozen=True)
@ -35,12 +41,14 @@ class PlaystationNetworkImageEntityDescription(ImageEntityDescription):
image_url_fn: Callable[[PlaystationNetworkData], str | None]
IMAGE_DESCRIPTIONS: tuple[PlaystationNetworkImageEntityDescription, ...] = (
IMAGE_DESCRIPTIONS_ME: tuple[PlaystationNetworkImageEntityDescription, ...] = (
PlaystationNetworkImageEntityDescription(
key=PlaystationNetworkImage.SHARE_PROFILE,
translation_key=PlaystationNetworkImage.SHARE_PROFILE,
image_url_fn=lambda data: data.shareable_profile_link["shareImageUrl"],
),
)
IMAGE_DESCRIPTIONS_ALL: tuple[PlaystationNetworkImageEntityDescription, ...] = (
PlaystationNetworkImageEntityDescription(
key=PlaystationNetworkImage.AVATAR,
translation_key=PlaystationNetworkImage.AVATAR,
@ -55,6 +63,14 @@ IMAGE_DESCRIPTIONS: tuple[PlaystationNetworkImageEntityDescription, ...] = (
)
),
),
PlaystationNetworkImageEntityDescription(
key=PlaystationNetworkImage.NOW_PLAYING_IMAGE,
translation_key=PlaystationNetworkImage.NOW_PLAYING_IMAGE,
image_url_fn=(
lambda data: get_game_title_info(data.presence).get("conceptIconUrl")
or get_game_title_info(data.presence).get("npTitleIconUrl")
),
),
)
@ -70,25 +86,43 @@ async def async_setup_entry(
async_add_entities(
[
PlaystationNetworkImageEntity(hass, coordinator, description)
for description in IMAGE_DESCRIPTIONS
for description in IMAGE_DESCRIPTIONS_ME + IMAGE_DESCRIPTIONS_ALL
]
)
for (
subentry_id,
friend_data_coordinator,
) in config_entry.runtime_data.friends.items():
async_add_entities(
[
PlaystationNetworkFriendImageEntity(
hass,
friend_data_coordinator,
description,
config_entry.subentries[subentry_id],
)
for description in IMAGE_DESCRIPTIONS_ALL
],
config_subentry_id=subentry_id,
)
class PlaystationNetworkImageEntity(PlaystationNetworkServiceEntity, ImageEntity):
class PlaystationNetworkImageBaseEntity(PlaystationNetworkServiceEntity, ImageEntity):
"""An image entity."""
entity_description: PlaystationNetworkImageEntityDescription
coordinator: PlaystationNetworkUserDataCoordinator
coordinator: PlayStationNetworkBaseCoordinator
def __init__(
self,
hass: HomeAssistant,
coordinator: PlaystationNetworkUserDataCoordinator,
coordinator: PlayStationNetworkBaseCoordinator,
entity_description: PlaystationNetworkImageEntityDescription,
subentry: ConfigSubentry | None = None,
) -> None:
"""Initialize the image entity."""
super().__init__(coordinator, entity_description)
super().__init__(coordinator, entity_description, subentry)
ImageEntity.__init__(self, hass)
self._attr_image_url = self.entity_description.image_url_fn(coordinator.data)
@ -96,6 +130,8 @@ class PlaystationNetworkImageEntity(PlaystationNetworkServiceEntity, ImageEntity
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
if TYPE_CHECKING:
assert isinstance(self.coordinator.data, PlaystationNetworkData)
url = self.entity_description.image_url_fn(self.coordinator.data)
if url != self._attr_image_url:
@ -104,3 +140,15 @@ class PlaystationNetworkImageEntity(PlaystationNetworkServiceEntity, ImageEntity
self._attr_image_last_updated = dt_util.utcnow()
super()._handle_coordinator_update()
class PlaystationNetworkImageEntity(PlaystationNetworkImageBaseEntity):
"""An image entity."""
coordinator: PlaystationNetworkUserDataCoordinator
class PlaystationNetworkFriendImageEntity(PlaystationNetworkImageBaseEntity):
"""An image entity."""
coordinator: PlaystationNetworkFriendDataCoordinator

View File

@ -19,11 +19,14 @@ from homeassistant.helpers.typing import StateType
from homeassistant.util import dt as dt_util
from .coordinator import (
PlayStationNetworkBaseCoordinator,
PlaystationNetworkConfigEntry,
PlaystationNetworkData,
PlaystationNetworkFriendDataCoordinator,
PlaystationNetworkUserDataCoordinator,
)
from .entity import PlaystationNetworkServiceEntity
from .helpers import get_game_title_info
PARALLEL_UPDATES = 0
@ -33,7 +36,6 @@ class PlaystationNetworkSensorEntityDescription(SensorEntityDescription):
"""PlayStation Network sensor description."""
value_fn: Callable[[PlaystationNetworkData], StateType | datetime]
entity_picture: str | None = None
available_fn: Callable[[PlaystationNetworkData], bool] = lambda _: True
@ -49,9 +51,10 @@ class PlaystationNetworkSensor(StrEnum):
ONLINE_ID = "online_id"
LAST_ONLINE = "last_online"
ONLINE_STATUS = "online_status"
NOW_PLAYING = "now_playing"
SENSOR_DESCRIPTIONS: tuple[PlaystationNetworkSensorEntityDescription, ...] = (
SENSOR_DESCRIPTIONS_TROPHY: tuple[PlaystationNetworkSensorEntityDescription, ...] = (
PlaystationNetworkSensorEntityDescription(
key=PlaystationNetworkSensor.TROPHY_LEVEL,
translation_key=PlaystationNetworkSensor.TROPHY_LEVEL,
@ -103,6 +106,8 @@ SENSOR_DESCRIPTIONS: tuple[PlaystationNetworkSensorEntityDescription, ...] = (
else None
),
),
)
SENSOR_DESCRIPTIONS_USER: tuple[PlaystationNetworkSensorEntityDescription, ...] = (
PlaystationNetworkSensorEntityDescription(
key=PlaystationNetworkSensor.ONLINE_ID,
translation_key=PlaystationNetworkSensor.ONLINE_ID,
@ -122,10 +127,19 @@ SENSOR_DESCRIPTIONS: tuple[PlaystationNetworkSensorEntityDescription, ...] = (
PlaystationNetworkSensorEntityDescription(
key=PlaystationNetworkSensor.ONLINE_STATUS,
translation_key=PlaystationNetworkSensor.ONLINE_STATUS,
value_fn=lambda psn: psn.availability.lower().replace("unavailable", "offline"),
value_fn=(
lambda psn: psn.presence["basicPresence"]["availability"]
.lower()
.replace("unavailable", "offline")
),
device_class=SensorDeviceClass.ENUM,
options=["offline", "availabletoplay", "availabletocommunicate", "busy"],
),
PlaystationNetworkSensorEntityDescription(
key=PlaystationNetworkSensor.NOW_PLAYING,
translation_key=PlaystationNetworkSensor.NOW_PLAYING,
value_fn=lambda psn: get_game_title_info(psn.presence).get("titleName"),
),
)
@ -138,18 +152,34 @@ async def async_setup_entry(
coordinator = config_entry.runtime_data.user_data
async_add_entities(
PlaystationNetworkSensorEntity(coordinator, description)
for description in SENSOR_DESCRIPTIONS
for description in SENSOR_DESCRIPTIONS_TROPHY + SENSOR_DESCRIPTIONS_USER
)
for (
subentry_id,
friend_data_coordinator,
) in config_entry.runtime_data.friends.items():
async_add_entities(
[
PlaystationNetworkFriendSensorEntity(
friend_data_coordinator,
description,
config_entry.subentries[subentry_id],
)
for description in SENSOR_DESCRIPTIONS_USER
],
config_subentry_id=subentry_id,
)
class PlaystationNetworkSensorEntity(
class PlaystationNetworkSensorBaseEntity(
PlaystationNetworkServiceEntity,
SensorEntity,
):
"""Representation of a PlayStation Network sensor entity."""
"""Base sensor entity."""
entity_description: PlaystationNetworkSensorEntityDescription
coordinator: PlaystationNetworkUserDataCoordinator
coordinator: PlayStationNetworkBaseCoordinator
@property
def native_value(self) -> StateType | datetime:
@ -169,14 +199,24 @@ class PlaystationNetworkSensorEntity(
(pic.get("url") for pic in profile_pictures if pic.get("size") == "xl"),
None,
)
return super().entity_picture
@property
def available(self) -> bool:
"""Return True if entity is available."""
return (
self.entity_description.available_fn(self.coordinator.data)
and super().available
return super().available and self.entity_description.available_fn(
self.coordinator.data
)
class PlaystationNetworkSensorEntity(PlaystationNetworkSensorBaseEntity):
"""Representation of a PlayStation Network sensor entity."""
coordinator: PlaystationNetworkUserDataCoordinator
class PlaystationNetworkFriendSensorEntity(PlaystationNetworkSensorBaseEntity):
"""Representation of a PlayStation Network sensor entity."""
coordinator: PlaystationNetworkFriendDataCoordinator

View File

@ -39,11 +39,40 @@
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"already_configured_as_subentry": "Already configured as a friend for another account. Delete the existing entry first.",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"unique_id_mismatch": "The provided NPSSO token corresponds to the account {wrong_account}. Please re-authenticate with the account **{name}**",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
}
},
"config_subentries": {
"friend": {
"step": {
"user": {
"title": "Friend online status",
"description": "Track the online status of a PlayStation Network friend.",
"data": {
"account_id": "Online ID"
},
"data_description": {
"account_id": "Select a friend from your friend list to track their online status."
}
}
},
"initiate_flow": {
"user": "Add friend"
},
"entry_type": "Friend",
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured_as_entry": "Already configured as a service. This account cannot be added as a friend.",
"already_configured": "Already configured as a friend in this or another account."
}
}
},
"exceptions": {
"not_ready": {
"message": "Authentication to the PlayStation Network failed."
@ -59,6 +88,12 @@
},
"send_message_failed": {
"message": "Failed to send message to group {group_name}. Try again later."
},
"user_profile_private": {
"message": "Unable to retrieve data for {user}. Privacy settings restrict access to activity."
},
"user_not_found": {
"message": "Unable to retrieve data for {user}. User does not exist or has been removed."
}
},
"entity": {
@ -104,6 +139,9 @@
"availabletocommunicate": "Online on PS App",
"busy": "Away"
}
},
"now_playing": {
"name": "Now playing"
}
},
"image": {
@ -112,6 +150,9 @@
},
"avatar": {
"name": "Avatar"
},
"now_playing_image": {
"name": "[%key:component::playstation_network::entity::sensor::now_playing::name%]"
}
},
"notify": {

View File

@ -4,6 +4,7 @@ from collections.abc import Generator
from datetime import UTC, datetime
from unittest.mock import AsyncMock, MagicMock, patch
from psnawp_api.models import User
from psnawp_api.models.group.group import Group
from psnawp_api.models.trophies import (
PlatformType,
@ -13,7 +14,12 @@ from psnawp_api.models.trophies import (
)
import pytest
from homeassistant.components.playstation_network.const import CONF_NPSSO, DOMAIN
from homeassistant.components.playstation_network.const import (
CONF_ACCOUNT_ID,
CONF_NPSSO,
DOMAIN,
)
from homeassistant.config_entries import ConfigSubentryData
from tests.common import MockConfigEntry
@ -32,6 +38,15 @@ def mock_config_entry() -> MockConfigEntry:
CONF_NPSSO: NPSSO_TOKEN,
},
unique_id=PSN_ID,
subentries_data=[
ConfigSubentryData(
data={CONF_ACCOUNT_ID: "fren-psn-id"},
subentry_id="ABCDEF",
subentry_type="friend",
title="PublicUniversalFriend",
unique_id="fren-psn-id",
)
],
)
@ -170,6 +185,12 @@ def mock_psnawpapi(mock_user: MagicMock) -> Generator[MagicMock]:
],
}
client.me.return_value.get_groups.return_value = [group]
fren = MagicMock(
spec=User, account_id="fren-psn-id", online_id="PublicUniversalFriend"
)
client.user.return_value.friends_list.return_value = [fren]
yield client

View File

@ -21,7 +21,6 @@
'title_name': "Assassin's Creed® III Liberation",
}),
}),
'availability': 'availableToPlay',
'presence': dict({
'basicPresence': dict({
'availability': 'availableToPlay',

View File

@ -146,6 +146,55 @@
'state': '2025-06-30T01:42:15+00:00',
})
# ---
# name: test_sensors[sensor.testuser_last_online_2-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.testuser_last_online_2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'Last online',
'platform': 'playstation_network',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': <PlaystationNetworkSensor.LAST_ONLINE: 'last_online'>,
'unique_id': 'fren-psn-id_last_online',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[sensor.testuser_last_online_2-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'timestamp',
'friendly_name': 'testuser Last online',
}),
'context': <ANY>,
'entity_id': 'sensor.testuser_last_online_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2025-06-30T01:42:15+00:00',
})
# ---
# name: test_sensors[sensor.testuser_next_level-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@ -195,6 +244,102 @@
'state': '19',
})
# ---
# name: test_sensors[sensor.testuser_now_playing-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.testuser_now_playing',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Now playing',
'platform': 'playstation_network',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': <PlaystationNetworkSensor.NOW_PLAYING: 'now_playing'>,
'unique_id': 'my-psn-id_now_playing',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[sensor.testuser_now_playing-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'testuser Now playing',
}),
'context': <ANY>,
'entity_id': 'sensor.testuser_now_playing',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'STAR WARS Jedi: Survivor™',
})
# ---
# name: test_sensors[sensor.testuser_now_playing_2-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.testuser_now_playing_2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Now playing',
'platform': 'playstation_network',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': <PlaystationNetworkSensor.NOW_PLAYING: 'now_playing'>,
'unique_id': 'fren-psn-id_now_playing',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[sensor.testuser_now_playing_2-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'testuser Now playing',
}),
'context': <ANY>,
'entity_id': 'sensor.testuser_now_playing_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'STAR WARS Jedi: Survivor™',
})
# ---
# name: test_sensors[sensor.testuser_online_id-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@ -244,6 +389,55 @@
'state': 'testuser',
})
# ---
# name: test_sensors[sensor.testuser_online_id_2-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.testuser_online_id_2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Online ID',
'platform': 'playstation_network',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': <PlaystationNetworkSensor.ONLINE_ID: 'online_id'>,
'unique_id': 'fren-psn-id_online_id',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[sensor.testuser_online_id_2-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'entity_picture': 'http://static-resource.np.community.playstation.net/avatar_xl/WWS_A/UP90001312L24_DD96EB6A4FF5FE883C09_XL.png',
'friendly_name': 'testuser Online ID',
}),
'context': <ANY>,
'entity_id': 'sensor.testuser_online_id_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'testuser',
})
# ---
# name: test_sensors[sensor.testuser_online_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@ -306,6 +500,68 @@
'state': 'availabletoplay',
})
# ---
# name: test_sensors[sensor.testuser_online_status_2-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'offline',
'availabletoplay',
'availabletocommunicate',
'busy',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.testuser_online_status_2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Online status',
'platform': 'playstation_network',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': <PlaystationNetworkSensor.ONLINE_STATUS: 'online_status'>,
'unique_id': 'fren-psn-id_online_status',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[sensor.testuser_online_status_2-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'testuser Online status',
'options': list([
'offline',
'availabletoplay',
'availabletocommunicate',
'busy',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.testuser_online_status_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'availabletoplay',
})
# ---
# name: test_sensors[sensor.testuser_platinum_trophies-entry]
EntityRegistryEntrySnapshot({
'aliases': set({

View File

@ -10,8 +10,17 @@ from homeassistant.components.playstation_network.config_flow import (
PSNAWPInvalidTokenError,
PSNAWPNotFoundError,
)
from homeassistant.components.playstation_network.const import CONF_NPSSO, DOMAIN
from homeassistant.config_entries import SOURCE_USER, ConfigEntryState
from homeassistant.components.playstation_network.const import (
CONF_ACCOUNT_ID,
CONF_NPSSO,
DOMAIN,
)
from homeassistant.config_entries import (
SOURCE_USER,
ConfigEntryState,
ConfigSubentry,
ConfigSubentryData,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@ -67,6 +76,45 @@ async def test_form_already_configured(
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("mock_psnawpapi")
async def test_form_already_configured_as_subentry(hass: HomeAssistant) -> None:
"""Test we abort form login when entry is already configured as subentry of another entry."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="PublicUniversalFriend",
data={
CONF_NPSSO: NPSSO_TOKEN,
},
unique_id="fren-psn-id",
subentries_data=[
ConfigSubentryData(
data={CONF_ACCOUNT_ID: PSN_ID},
subentry_id="ABCDEF",
subentry_type="friend",
title="test-user",
unique_id=PSN_ID,
)
],
)
config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_NPSSO: NPSSO_TOKEN},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured_as_subentry"
@pytest.mark.parametrize(
("raise_error", "text_error"),
[
@ -325,3 +373,123 @@ async def test_flow_reconfigure(
assert config_entry.data[CONF_NPSSO] == "NEW_NPSSO_TOKEN"
assert len(hass.config_entries.async_entries()) == 1
@pytest.mark.usefixtures("mock_psnawpapi")
async def test_add_friend_flow(hass: HomeAssistant) -> None:
"""Test add friend subentry flow."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="test-user",
data={
CONF_NPSSO: NPSSO_TOKEN,
},
unique_id=PSN_ID,
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
result = await hass.config_entries.subentries.async_init(
(config_entry.entry_id, "friend"),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={CONF_ACCOUNT_ID: "fren-psn-id"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
subentry_id = list(config_entry.subentries)[0]
assert config_entry.subentries == {
subentry_id: ConfigSubentry(
data={},
subentry_id=subentry_id,
subentry_type="friend",
title="PublicUniversalFriend",
unique_id="fren-psn-id",
)
}
@pytest.mark.usefixtures("mock_psnawpapi")
async def test_add_friend_flow_already_configured(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
"""Test we abort add friend subentry flow when already configured."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
result = await hass.config_entries.subentries.async_init(
(config_entry.entry_id, "friend"),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={CONF_ACCOUNT_ID: "fren-psn-id"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("mock_psnawpapi")
async def test_add_friend_flow_already_configured_as_entry(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
"""Test we abort add friend subentry flow when already configured as config entry."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="test-user",
data={
CONF_NPSSO: NPSSO_TOKEN,
},
unique_id=PSN_ID,
)
fren_config_entry = MockConfigEntry(
domain=DOMAIN,
title="PublicUniversalFriend",
data={
CONF_NPSSO: NPSSO_TOKEN,
},
unique_id="fren-psn-id",
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
fren_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(fren_config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
result = await hass.config_entries.subentries.async_init(
(config_entry.entry_id, "friend"),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={CONF_ACCOUNT_ID: "fren-psn-id"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured_as_entry"

View File

@ -7,6 +7,7 @@ from freezegun.api import FrozenDateTimeFactory
from psnawp_api.core import (
PSNAWPAuthenticationError,
PSNAWPClientError,
PSNAWPForbiddenError,
PSNAWPNotFoundError,
PSNAWPServerError,
)
@ -263,3 +264,83 @@ async def test_trophy_title_coordinator_play_new_game(
state.attributes["entity_picture"]
== "https://image.api.playstation.com/trophy/np/NPWR03134_00_0008206095F67FD3BB385E9E00A7C9CFE6F5A4AB96/5F87A6997DD23D1C4D4CC0D1F958ED79CB905331.PNG"
)
@pytest.mark.parametrize(
"exception",
[PSNAWPNotFoundError, PSNAWPServerError, PSNAWPClientError, PSNAWPForbiddenError],
)
async def test_friends_coordinator_update_data_failed(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_psnawpapi: MagicMock,
exception: Exception,
) -> None:
"""Test friends coordinator setup fails in _update_data."""
mock_psnawpapi.user.return_value.get_presence.side_effect = [
mock_psnawpapi.user.return_value.get_presence.return_value,
exception,
]
config_entry.add_to_hass(hass)
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(
("exception", "state"),
[
(PSNAWPNotFoundError, ConfigEntryState.SETUP_ERROR),
(PSNAWPAuthenticationError, ConfigEntryState.SETUP_ERROR),
(PSNAWPServerError, ConfigEntryState.SETUP_RETRY),
(PSNAWPClientError, ConfigEntryState.SETUP_RETRY),
],
)
async def test_friends_coordinator_setup_failed(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_psnawpapi: MagicMock,
exception: Exception,
state: ConfigEntryState,
) -> None:
"""Test friends coordinator setup fails in _async_setup."""
mock_psnawpapi.user.side_effect = [
mock_psnawpapi.user.return_value,
exception,
]
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is state
async def test_friends_coordinator_auth_failed(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_psnawpapi: MagicMock,
) -> None:
"""Test friends coordinator starts reauth on authentication error."""
mock_psnawpapi.user.side_effect = [
mock_psnawpapi.user.return_value,
PSNAWPAuthenticationError,
]
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_ERROR
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
flow = flows[0]
assert flow.get("step_id") == "reauth_confirm"
assert flow.get("handler") == DOMAIN
assert "context" in flow
assert flow["context"].get("source") == SOURCE_REAUTH
assert flow["context"].get("entry_id") == config_entry.entry_id